test: Cover visitDifferences node skipping

visitDifferences() is reached in production only through a file-static
populateFetchPack() in LedgerMaster.cpp, so the private hasLeafNode()
and hasInnerNode() it consults were untested. Four cases pin both
answers: only the nodes on a missing item's path, nothing against an
identical map, every node against a null map, and a callback stop.

The first case uses 64 shared items, so the shared leaves build inner
nodes of their own and whole matching subtrees exist to skip. A
finalize() helper seals each map first, since visitDifferences() returns
early while the root hash is zero, and getHash() is what seals it, by
unsharing such a map. Each call is wrapped in ASSERT_NO_FATAL_FAILURE,
which carries a gtest ASSERT_ failure out of a helper to its caller.

.cspell.config.yaml

@@ -103,6 +103,7 @@ words:
   - dearmor
   - decryptor
   - dedented
+  - dedup
   - deleteme
   - demultiplexer
   - deserializaton
@@ -344,6 +345,7 @@ words:
   - unambiguity
   - unauthorizes
   - unauthorizing
+  - undeserializable
   - unergonomic
   - unfetched
   - unfindable

.github/scripts/levelization/results/ordering.txt

@@ -62,6 +62,7 @@ libxrpl.tx > xrpl.protocol
 libxrpl.tx > xrpl.server
 libxrpl.tx > xrpl.tx
 test.app > test.jtx
+test.app > tests.libxrpl
 test.app > test.unit_test
 test.app > xrpl.basics
 test.app > xrpl.config

cmake/XrplSanitizers.cmake

@@ -77,19 +77,24 @@ if(is_clang)
     message(STATUS "  Ignorelist: ${ignorelist_path}")
 endif()
 
-# Define SANITIZERS macro for BuildInfo.cpp
+# Define the SANITIZERS macro for BuildInfo.cpp, plus one of XRPL_ASAN,
+# XRPL_TSAN and XRPL_UBSAN per active sanitizer, so that code can test for a
+# specific one with #ifdef instead of parsing the dot-joined SANITIZERS string.
 set(sanitizers_list)
 if(SANITIZERS MATCHES "address")
     set(enable_asan ON)
     list(APPEND sanitizers_list "ASAN")
+    target_compile_definitions(common INTERFACE XRPL_ASAN)
 endif()
 if(SANITIZERS MATCHES "thread")
     set(enable_tsan ON)
     list(APPEND sanitizers_list "TSAN")
+    target_compile_definitions(common INTERFACE XRPL_TSAN)
 endif()
 if(SANITIZERS MATCHES "undefinedbehavior")
     set(enable_ubsan ON)
     list(APPEND sanitizers_list "UBSAN")
+    target_compile_definitions(common INTERFACE XRPL_UBSAN)
 endif()
 
 if(sanitizers_list)

include/xrpl/basics/IntrusivePointer.h

@@ -86,12 +86,15 @@ public:
         requires std::convertible_to<TT*, T*>
     SharedIntrusive(SharedIntrusive<TT> const& rhs);
 
-    SharedIntrusive(SharedIntrusive&& rhs);
+    // noexcept so move_if_noexcept moves rather than copies, which NodePathStack's path_ vector
+    // relies on when it relocates. The body is a std::exchange on a raw pointer, which is what
+    // makes the noexcept honest.
+    SharedIntrusive(SharedIntrusive&& rhs) noexcept;
 
     template <class TT>
         requires std::convertible_to<TT*, T*>
-    SharedIntrusive(
-        SharedIntrusive<TT>&& rhs);  // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved)
+    // NOLINTNEXTLINE(cppcoreguidelines-rvalue-reference-param-not-moved)
+    SharedIntrusive(SharedIntrusive<TT>&& rhs) noexcept;
 
     SharedIntrusive&
     operator=(SharedIntrusive const& rhs);
@@ -529,6 +532,22 @@ staticPointerCast(TT const& v)
     return SharedPtr<T>(StaticCastTagSharedIntrusive{}, v);
 }
 
+/**
+ * Statically cast an intrusive pointer, moving out of it.
+ *
+ * Spelling the parameter as `SharedIntrusive<TT>&&` binds rvalues only and
+ * leaves lvalues to the `const&` overload above.
+ *
+ * @param v the pointer to cast, left empty afterwards.
+ * @return a pointer of the requested type to the same object.
+ */
+template <class T, class TT>
+[[nodiscard]] SharedPtr<T>
+staticPointerCast(SharedIntrusive<TT>&& v)
+{
+    return SharedPtr<T>(StaticCastTagSharedIntrusive{}, std::move(v));
+}
+
 template <class T, class TT>
 SharedPtr<T>
 dynamicPointerCast(TT const& v)

include/xrpl/basics/IntrusivePointer.ipp

@@ -43,7 +43,7 @@ SharedIntrusive<T>::SharedIntrusive(SharedIntrusive<TT> const& rhs)
 }
 
 template <class T>
-SharedIntrusive<T>::SharedIntrusive(SharedIntrusive&& rhs)
+SharedIntrusive<T>::SharedIntrusive(SharedIntrusive&& rhs) noexcept
     : ptr_{std::move(rhs).unsafeExchange(nullptr)}
 {
 }
@@ -51,7 +51,7 @@ SharedIntrusive<T>::SharedIntrusive(SharedIntrusive&& rhs)
 template <class T>
 template <class TT>
     requires std::convertible_to<TT*, T*>
-SharedIntrusive<T>::SharedIntrusive(SharedIntrusive<TT>&& rhs)
+SharedIntrusive<T>::SharedIntrusive(SharedIntrusive<TT>&& rhs) noexcept
     : ptr_{std::move(rhs).unsafeExchange(nullptr)}
 {
 }

include/xrpl/basics/KeyCache.h

@@ -1,10 +1,15 @@
 #pragma once
 
 #include <xrpl/basics/TaggedCache.h>
-#include <xrpl/basics/base_uint.h>
 
 namespace xrpl {
 
-using KeyCache = TaggedCache<UInt256, int, true>;
+/**
+ * TaggedCache in key-only mode, holding no value per key.
+ *
+ * @tparam Key the key type to remember.
+ */
+template <class Key>
+using KeyCache = TaggedCache<Key, int, true>;
 
 }  // namespace xrpl

include/xrpl/ledger/Ledger.h

@@ -105,6 +105,16 @@ public:
         std::vector<UInt256> const& amendments,
         Family& family);
 
+    /**
+     * Create a ledger from a header whose maps are filled in afterwards.
+     *
+     * Both maps start Synching against the hashes the header carries, which are
+     * input rather than derived, so setImmutable() leaves them alone.
+     *
+     * @param info The header to build from.
+     * @param rules The rules in force.
+     * @param family The SHAMap family the maps belong to.
+     */
     Ledger(LedgerHeader const& info, Rules rules, Family& family);
 
     /**
@@ -257,38 +267,79 @@ public:
         header_.validated = true;
     }
 
-    void
+    /**
+     * Mark this ledger as accepted and attempt to make it immutable.
+     *
+     * The close-time fields are recorded first, since the ledger hash covers
+     * them.
+     *
+     * @param closeTime The consensus-agreed close time.
+     * @param closeResolution The close time resolution.
+     * @param correctCloseTime Whether consensus agreed on the close time. If
+     *        false, kSLcfNoConsensusTime is recorded in closeFlags instead.
+     * @return What setImmutable() returned.
+     */
+    [[nodiscard]] bool
     setAccepted(
         NetClock::time_point closeTime,
         NetClock::duration closeResolution,
         bool correctCloseTime);
 
-    void
+    /**
+     * Mark this ledger as immutable, so it can no longer be modified.
+     *
+     * Only a map syncing against hashes from outside can be found unsound (see
+     * SHAMap::addKnownNode), so a caller whose ledger was built or loaded
+     * locally treats a false return as a broken internal invariant.
+     *
+     * @param rehash Whether to recompute the ledger hash from the header
+     *        fields. The transaction and account hashes are recomputed from the
+     *        maps too, but only the first time and only if the header did not
+     *        supply them.
+     * @return false if either map is Invalid, leaving the immutable flag as it
+     *         was and the header untouched. The ledger must then be discarded
+     *         rather than retried, since one map may already be immutable.
+     */
+    [[nodiscard]] bool
     setImmutable(bool rehash = true);
 
+    /**
+     * @return Whether both maps have been settled, so the ledger can no longer
+     *         change.
+     */
     bool
     isImmutable() const
     {
         return immutable_;
     }
 
-    /*  Mark this ledger as "should be full".
+    /**
+     * @return Whether both maps can still be the maps the header names. See
+     *         SHAMap::isValid().
+     */
+    [[nodiscard]] bool
+    mapsValid() const
+    {
+        return txMap_.isValid() && stateMap_.isValid();
+    }
 
-        "Full" is metadata property of the ledger, it indicates
-        that the local server wants all the corresponding nodes
-        in durable storage.
-
-        This is marked `const` because it reflects metadata
-        and not data that is in common with other nodes on the
-        network.
-    */
+    /**
+     * Mark this ledger as "should be full", indicating that the local server
+     * wants all the corresponding nodes in durable storage.
+     *
+     * Const because it reflects metadata, not data this ledger shares with
+     * other nodes on the network.
+     */
     void
     setFull() const
     {
-        txMap_.setFull();
+        // Sequence before flag, per map: setLedgerSeq() stores relaxed and setFull() stores
+        // release, and SHAMap::finishFetch() reads the sequence only after acquiring the flag, so
+        // only this order publishes it.
         txMap_.setLedgerSeq(header_.seq);
-        stateMap_.setFull();
+        txMap_.setFull();
         stateMap_.setLedgerSeq(header_.seq);
+        stateMap_.setFull();
     }
 
     void
@@ -418,8 +469,33 @@ private:
     static std::pair<std::shared_ptr<STTx const>, std::shared_ptr<STObject const>>
     deserializeTxPlusMeta(SHAMapItem const& item);
 
+    /**
+     * Make both maps immutable.
+     *
+     * Each map is settled on its own, so neither is left mid-sync because the
+     * other refused.
+     *
+     * @return Whether both maps are immutable.
+     */
+    [[nodiscard]] bool
+    setMapsImmutable()
+    {
+        bool const txImmutable = txMap_.setImmutable();
+        bool const stateImmutable = stateMap_.setImmutable();
+        return txImmutable && stateImmutable;
+    }
+
     bool immutable_;
 
+    /**
+     * Whether the header supplied the transaction and account hashes, so they
+     * must not be derived from the maps.
+     *
+     * True only for a ledger built from a header, which stays verified against
+     * the hash it was asked for. Fixed at construction, unlike immutable_.
+     */
+    bool const mapHashesFromHeader_ = false;
+
     // A SHAMap containing the transactions associated with this ledger.
     SHAMap mutable txMap_;
 

include/xrpl/shamap/FullBelowCache.h

@@ -3,9 +3,12 @@
 #include <xrpl/basics/KeyCache.h>
 #include <xrpl/basics/TaggedCache.h>
 #include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/partitioned_unordered_map.h>
+#include <xrpl/beast/hash/hash_append.h>
 #include <xrpl/beast/insight/Collector.h>
 #include <xrpl/beast/insight/NullCollector.h>
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/shamap/SHAMapNodeID.h>
 
 #include <atomic>
 #include <chrono>
@@ -15,6 +18,112 @@
 
 namespace xrpl {
 
+/**
+ * Names a subtree whose descendants are all resident, by hash and by position.
+ *
+ * A node's hash covers its children but neither its depth nor its path, and one
+ * cache answers for every map built from a NodeFamily, so the position is what
+ * makes a hit answer for the position asked about and no other. The depth is
+ * kept beside the id because SHAMapNodeID masks its id to its own depth, so an
+ * all-zero id belongs to the root and to every all-zero prefix alike.
+ *
+ * Held as two fields rather than as a SHAMapNodeID, which is a CountedObject:
+ * one per entry would report the cache's population as live node IDs in the
+ * get_counts RPC, and would put an atomic increment on every temporary key a
+ * lookup builds.
+ */
+struct FullBelowKey
+{
+    /**
+     * The hash of the subtree.
+     */
+    UInt256 hash;
+
+    /**
+     * The masked id of the node the subtree was completed at.
+     */
+    UInt256 position;
+
+    /**
+     * The depth of that node.
+     */
+    std::uint32_t depth{0};
+
+    /**
+     * @param subtreeHash The hash of the subtree whose descendants are all
+     *        resident.
+     * @param nodeID The node the subtree was completed at.
+     */
+    FullBelowKey(UInt256 const& subtreeHash, SHAMapNodeID const& nodeID)
+        : hash(subtreeHash)
+        , position(nodeID.getNodeID())
+        , depth(static_cast<std::uint32_t>(nodeID.getDepth()))
+    {
+    }
+
+    /**
+     * The same key, named by the node above the position and the branch taken.
+     *
+     * Produces the fields the constructor above would for that child's own node
+     * ID, so the two forms name one entry.
+     *
+     * @param subtreeHash The hash of the subtree that is fully resident.
+     * @param parentID The node above the one the subtree was completed at.
+     * @param branch The branch of that node leading to it.
+     */
+    FullBelowKey(UInt256 const& subtreeHash, SHAMapNodeID const& parentID, unsigned int branch)
+        : hash(subtreeHash)
+        , position(childNodeID(parentID.getNodeID(), parentID.getDepth(), branch))
+        , depth(static_cast<std::uint32_t>(parentID.getDepth() + 1))
+    {
+    }
+
+    [[nodiscard]] bool
+    operator==(FullBelowKey const& other) const = default;
+};
+
+// The per-entry cost of this cache is computed from these sizes, so pin the layout that arithmetic
+// assumes: the three fields sit back to back with no padding.
+static_assert(
+    sizeof(FullBelowKey) == (2 * sizeof(UInt256)) + sizeof(std::uint32_t),
+    "FullBelowKey must stay free of padding");
+static_assert(
+    alignof(FullBelowKey) == alignof(std::uint32_t),
+    "FullBelowKey's lack of padding rests on its alignment");
+
+/**
+ * Feed a key to a hasher, field by field.
+ *
+ * Written out rather than hashed in one call over the object's bytes, since
+ * beast::IsUniquelyRepresented is opt-in per type and has no case for a
+ * user-defined struct, whatever its layout.
+ *
+ * @param h The hasher to feed.
+ * @param key The key to hash.
+ */
+template <class Hasher>
+void
+hash_append(Hasher& h, FullBelowKey const& key) noexcept
+{
+    using beast::hash_append;
+    hash_append(h, key.hash, key.position, key.depth);
+}
+
+/**
+ * Choose which partition of the cache's map a key lives in.
+ *
+ * Taken from the subtree hash, which is already uniformly distributed.
+ *
+ * @param key The key to place.
+ * @return The value PartitionedUnorderedMap reduces by its partition count.
+ */
+template <>
+inline std::size_t
+extract(FullBelowKey const& key)
+{
+    return extract(key.hash);
+}
+
 namespace detail {
 
 /**
@@ -24,12 +133,12 @@ namespace detail {
 class BasicFullBelowCache
 {
 private:
-    using CacheType = KeyCache;
+    using CacheType = KeyCache<FullBelowKey>;
 
 public:
     static constexpr auto kDefaultCacheTargetSize = 0;
 
-    using key_type = UInt256;
+    using key_type = FullBelowKey;
     using ClockType = CacheType::ClockType;
 
     /**
@@ -86,13 +195,15 @@ public:
      * Refresh the last access time of an item, if it exists.
      * Thread safety:
      *     Safe to call from any thread.
-     * @param key The key to refresh.
-     * @return `true` If the key exists.
+     * @param hash The hash of the subtree to ask about.
+     * @param parentID The node above the position to ask about it at.
+     * @param branch The branch of that node leading to that position.
+     * @return `true` If that subtree is recorded as resident at that position.
      */
     bool
-    touchIfExists(key_type const& key)
+    touchIfExists(UInt256 const& hash, SHAMapNodeID const& parentID, unsigned int branch)
     {
-        return cache_.touchIfExists(key);
+        return cache_.touchIfExists(key_type{hash, parentID, branch});
     }
 
     /**
@@ -101,12 +212,13 @@ public:
      * be refreshed.
      * Thread safety:
      *     Safe to call from any thread.
-     * @param key The key to insert.
+     * @param hash The hash of the subtree whose descendants are all resident.
+     * @param nodeID The node that subtree was completed at.
      */
     void
-    insert(key_type const& key)
+    insert(UInt256 const& hash, SHAMapNodeID const& nodeID)
     {
-        cache_.insert(key);
+        cache_.insert(key_type{hash, nodeID});
     }
 
     /**

include/xrpl/shamap/README.md

@@ -278,11 +278,14 @@ This cache remembers which trie keys have all of their children resident in a
 when creating the missing nodes list. Missing nodes are those nodes that a
 `SHAMap` refers to but that are not stored in the local database.
 
-As a depth-first walk of a `SHAMap` is performed, if an inner node answers true to
-`isFullBelow()` then it is known that none of this node's children are missing
-nodes, and thus that subtree does not need to be walked. These nodes are stored
-in the FullBelowCache. Subsequent walks check the FullBelowCache first when
-encountering a node, and ignore that subtree if found.
+As a depth-first walk of a `SHAMap` completes an inner node, meaning none of the
+nodes below it are missing, the node's hash and position are stored in the
+FullBelowCache. Subsequent walks check the FullBelowCache when encountering a
+child, and skip that subtree if found. The entry is keyed by position as well as
+hash, since a hash commits to a node's contents rather than its place, and the
+same node object is shared by hash between maps and positions. For the same
+reason the inner node's own `isFullBelow()` flag is written and read for a
+walk's root alone, where the position is fixed.
 
 ## `SHAMapTreeNode`
 

include/xrpl/shamap/SHAMap.h

@@ -2,6 +2,7 @@
 
 #include <xrpl/basics/Blob.h>
 #include <xrpl/basics/IntrusivePointer.h>
+#include <xrpl/basics/Log.h>
 #include <xrpl/basics/SHAMapHash.h>
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/beast/utility/Journal.h>
@@ -17,6 +18,7 @@
 #include <xrpl/shamap/SHAMapNodeID.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
+#include <atomic>
 #include <condition_variable>
 #include <cstddef>
 #include <cstdint>
@@ -30,6 +32,7 @@
 #include <set>
 #include <stack>
 #include <tuple>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -40,7 +43,7 @@ class SHAMapSyncFilter;
 /**
  * Describes the current state of a given SHAMap
  */
-enum class SHAMapState {
+enum class SHAMapState : std::uint8_t {
     /**
      * The map is in flux and objects can be added and removed.
      *
@@ -71,20 +74,21 @@ enum class SHAMapState {
 };
 
 /**
- * A SHAMap is both a radix tree with a fan-out of 16 and a Merkle tree.
+ * A SHAMap is both a trie with a fan-out of 16 and a Merkle tree.
  *
- * A radix tree is a tree with two properties:
+ * A trie holds a key in the position of its nodes: the path from the root down
+ * to a node spells out the prefix every key below it shares (the "prefix
+ * property"). A SHAMap spends one nibble of the 256-bit key per level, so an
+ * inner node has at most 16 children and a leaf sits at depth 64 at the
+ * deepest. A leaf also carries its own full key, so a reader can check that it
+ * was reached through the branches that key names.
  *
- *   1. The key for a node is represented by the node's position in the tree
- *      (the "prefix property").
- *   2. A node with only one child is merged with that child
- *      (the "merge property")
+ * An insert creates an inner node at every nibble two keys share, and a delete
+ * collapses a chain that reduces to one leaf. Every edge spans exactly one
+ * nibble, so a path has one entry per level and 65 entries at the most, and a
+ * path's length names each node's depth. Traversal relies on that.
  *
- * These properties result in a significantly smaller memory footprint for
- * a radix tree.
- *
- * A fan-out of 16 means that each node in the tree has at most 16
- * children. See https://en.wikipedia.org/wiki/Radix_tree
+ * See https://en.wikipedia.org/wiki/Trie
  *
  * A Merkle tree is a tree where each non-leaf node is labelled with the hash
  * of the combined labels of its children nodes.
@@ -120,21 +124,43 @@ private:
      */
     std::uint32_t cowid_ = 1;
 
+    // ledgerSeq_, state_ and full_ are touched on the nodestore fetch path and on a
+    // getMissingNodes() walk, neither of which may block, so pin them lock-free.
+    static_assert(std::atomic<std::uint32_t>::is_always_lock_free);
+    static_assert(std::atomic<SHAMapState>::is_always_lock_free);
+    static_assert(std::atomic<bool>::is_always_lock_free);
+
     /**
      * The sequence of the ledger that this map references, if any.
+     *
+     * Written by Ledger::setFull() and by InboundLedger, and read on a
+     * nodestore reader thread. Relaxed both ways, since it is only a lookup
+     * hint for a store keyed by hash.
      */
-    std::uint32_t ledgerSeq_ = 0;
+    std::atomic<std::uint32_t> ledgerSeq_ = 0;
 
     SHAMapTreeNodePtr root_;
-    mutable SHAMapState state_;
+
+    /**
+     * The map's state. Atomic because a getMissingNodes() walk writes it with
+     * the acquisition's lock released. Mutable because a const descend() can
+     * record the Invalid verdict.
+     */
+    mutable std::atomic<SHAMapState> state_;
     SHAMapType const type_;
-    bool backed_ = true;         // Map is backed by the database
-    mutable bool full_ = false;  // Map is believed complete in database
+    bool backed_ = true;  // Map is backed by the database
+
+    /**
+     * Map is believed complete in database.
+     *
+     * Atomic because finishFetch() clears it on any nodestore reader thread,
+     * several of which can run at once.
+     */
+    mutable std::atomic<bool> full_ = false;
 
 public:
     /**
-     * Number of children each non-leaf node has (the 'radix tree' part of the
-     * map)
+     * Number of children each non-leaf node has, which is the trie's fan-out
      */
     static constexpr unsigned int kBranchFactor = SHAMapInnerNode::kBranchFactor;
 
@@ -143,6 +169,18 @@ public:
      */
     static constexpr unsigned int kLeafDepth = 64;
 
+    /**
+     * Whether only a leaf may occupy a position at `depth`.
+     *
+     * @param depth the depth to judge.
+     * @return whether that depth is at or past kLeafDepth.
+     */
+    [[nodiscard]] static constexpr bool
+    isLeafDepth(unsigned int depth)
+    {
+        return depth >= kLeafDepth;
+    }
+
     using DeltaItem =
         std::pair<boost::intrusive_ptr<SHAMapItem const>, boost::intrusive_ptr<SHAMapItem const>>;
     using Delta = std::map<UInt256, DeltaItem>;
@@ -152,7 +190,14 @@ public:
     SHAMap&
     operator=(SHAMap const&) = delete;
 
-    // Take a snapshot of the given map:
+    /**
+     * Take a snapshot of the given map.
+     *
+     * @param other The map to snapshot. An Invalid source yields an Invalid
+     *        snapshot.
+     * @param isMutable Whether the snapshot may be modified. Ignored when other
+     *        is Invalid.
+     */
     SHAMap(SHAMap const& other, bool isMutable);
 
     // build new map
@@ -190,8 +235,15 @@ public:
 
     //--------------------------------------------------------------------------
 
-    // Returns a new map that's a snapshot of this one.
-    // Handles copy on write for mutable snapshots.
+    /**
+     * Return a new map that is a snapshot of this one.
+     *
+     * Handles copy on write for mutable snapshots. An invalid map yields an
+     * invalid snapshot.
+     *
+     * @param isMutable Whether the snapshot may be modified.
+     * @return The snapshot.
+     */
     std::shared_ptr<SHAMap>
     snapShot(bool isMutable) const;
 
@@ -243,9 +295,11 @@ public:
     /**
      * Find the first item after the given item.
      *
-     * @param id the identifier of the item.
-     *
-     * @note The item does not need to exist.
+     * @param id the identifier of the item, which need not exist in the map.
+     * @return an iterator at the first item with a greater key, or end() if the
+     *         map holds no greater key.
+     * @throws SHAMapMissingNode if the map cannot be walked. end() is the
+     *         separate answer that the map holds no greater key.
      */
     ConstIterator
     upperBound(UInt256 const& id) const;
@@ -253,9 +307,11 @@ public:
     /**
      * Find the object with the greatest object id smaller than the input id.
      *
-     * @param id the identifier of the item.
-     *
-     * @note The item does not need to exist.
+     * @param id the identifier of the item, which need not exist in the map.
+     * @return an iterator at the last item with a smaller key, or end() if the
+     *         map holds no smaller key.
+     * @throws SHAMapMissingNode if the map cannot be walked. end() is the
+     *         separate answer that the map holds no smaller key.
      */
     ConstIterator
     lowerBound(UInt256 const& id) const;
@@ -296,9 +352,13 @@ public:
      * concurrency, to discover nodes referenced in the
      * SHAMap but not available locally.
      *
+     * Only a leaf may occupy a position at or beyond kLeafDepth. A map that
+     * breaks that is marked Invalid and the traversal is abandoned, so callers
+     * ask isValid() to tell an empty result from a satisfied map.
+     *
      * @param maxNodes The maximum number of found nodes to return
      * @param filter The filter to use when retrieving nodes
-     * @param return The nodes known to be missing
+     * @return The nodes known to be missing, or empty if the map is Invalid
      */
     std::vector<std::pair<SHAMapNodeID, UInt256>>
     getMissingNodes(int maxNodes, SHAMapSyncFilter const* filter);
@@ -341,6 +401,9 @@ public:
      * This function is used when receiving the root node of a SHAMap from a peer during ledger
      * synchronization. The node must already have been deserialized.
      *
+     * A root offered under a hash the map does not hold names another tree and
+     * is reported as invalid data. A root matching that hash is a duplicate.
+     *
      * @param hash The expected hash of the root node.
      * @param rootNode A deserialized root node to add.
      * @param filter Optional sync filter to track received nodes.
@@ -360,14 +423,26 @@ public:
      * is inserted at the position specified by nodeID. The node must already have been
      * deserialized.
      *
+     * A node that no valid tree can hold makes the map Invalid, which is
+     * terminal. An acquisition reaching that verdict gives up.
+     *
      * @param nodeID The position in the tree where this node belongs.
      * @param treeNode A deserialized tree node to add.
      * @param filter Optional sync filter to track received nodes.
      * @return Status indicating whether the node was useful, duplicate, or invalid.
      *
-     * @note This function expects the treeNode to be a valid, deserialized SHAMapTreeNode. The
-     *       caller is responsible for deserialization and basic validation before calling this
-     *       function. This also means that the nodeID must be consistent with the node's content.
+     * @note The caller is responsible for deserialization. The position is
+     *       checked here, in three steps that share two verdicts. A node
+     *       the descent reaches at a position other than the one nodeID
+     *       claims is refused with invalid(), and the map stays valid, since
+     *       only the label was wrong. A leaf that hash-verified at the
+     *       position nodeID claims, but whose own key does not lie under
+     *       nodeID, condemns the map and returns mapInvalidated(). A node the
+     *       descent itself condemns on the way, resolved from the local store
+     *       or the filter rather than supplied by the caller, leaves the map
+     *       Invalid and returns invalid(), since that verdict is not the
+     *       caller's doing. Any map outside Synching, including one already
+     *       condemned, returns duplicate().
      */
     SHAMapAddNode
     addKnownNode(
@@ -375,16 +450,57 @@ public:
         SHAMapTreeNodePtr treeNode,
         SHAMapSyncFilter const* filter);
 
-    // status functions
-    void
+    /**
+     * Mark this map as immutable, so it can no longer be modified.
+     *
+     * @return false if the map is Invalid and was left unchanged, true
+     *         otherwise.
+     */
+    [[nodiscard]] bool
     setImmutable();
-    bool
+
+    /**
+     * @return Whether the map is being synced against a hash it was given, so
+     *         its hash is fixed while nodes may still be added.
+     */
+    [[nodiscard]] bool
     isSynching() const;
+
+    /**
+     * @return Whether the map is settled, with its hash and nodes fixed.
+     */
+    [[nodiscard]] bool
+    isImmutable() const;
+
+    /**
+     * Mark this map as syncing, fixing its hash while still allowing missing
+     * nodes to be added.
+     *
+     * The map must be freshly constructed. The body asserts that the map is not
+     * Invalid, so a caller that ignores the precondition stops a build with
+     * assertions enabled.
+     */
     void
     setSynching();
+
+    /**
+     * Mark this map as no longer syncing, so it can be modified again.
+     *
+     * Moves only a Synching map. An Immutable map stays settled, a Modifying
+     * map has nothing to clear, and Invalid is terminal.
+     */
     void
     clearSynching();
-    bool
+
+    /**
+     * Whether the map can still be the map it claims to be.
+     *
+     * A map that is merely missing nodes is valid, and stays valid until
+     * something proves the tree it is syncing against cannot exist.
+     *
+     * @return Whether the map has not been proven impossible.
+     */
+    [[nodiscard]] bool
     isValid() const;
 
     // caution: otherMap must be accessed only by this function
@@ -421,109 +537,362 @@ public:
 
 private:
     /**
-     * A path from the root of the map down to some node, pairing each node with the ID naming its
-     * position.
+     * Whether placing `node` one level below `parentDepth` leaves it no room.
      *
-     * The two halves of an entry must agree, and the only way to get that wrong is to compute an ID
-     * from the wrong branch. So this type does not accept an ID at all: every push takes the branch
-     * being descended and derives the ID itself, so a node and its ID cannot disagree. Reads are
-     * exposed through the same accessors a std::stack would offer.
+     * Only a leaf may sit at kLeafDepth, and only a parent above kLeafDepth has
+     * a nibble left to record the branch taken. Both bounds are needed, since a
+     * leaf one level above kLeafDepth is allowed. Position is judged separately,
+     * by the walk that keeps a path.
+     *
+     * The conjunct order is load bearing: the depth is tested first, so the
+     * virtual isInner() call runs only at the one level where the bound
+     * applies.
+     *
+     * @param parentDepth the depth of the node being descended from.
+     * @param node the node about to be placed one level below it.
+     * @return whether that placement is past the deepest level this kind of
+     *         node may occupy.
+     */
+    [[nodiscard]] static bool
+    pastLeafDepth(unsigned int parentDepth, SHAMapTreeNode const& node)
+    {
+        return isLeafDepth(parentDepth + 1u) && (node.isInner() || isLeafDepth(parentDepth));
+    }
+
+    /**
+     * A root-down path of nodes through the map.
+     *
+     * Entry `i` is the node at depth `i`, since every edge spans one nibble
+     * (see the class docstring). Only the node is stored. A caller that needs a
+     * position reads a depth and takes the nibbles from its own key.
      */
     class NodePathStack
     {
     public:
+        /**
+         * @return whether the path holds no node at all.
+         */
         [[nodiscard]] bool
         empty() const
         {
-            return stack_.empty();
+            return path_.empty();
         }
 
+        /**
+         * @return how many nodes the path holds, one more than the last node's
+         *         depth.
+         */
         [[nodiscard]] std::size_t
         size() const
         {
-            return stack_.size();
+            return path_.size();
         }
 
-        [[nodiscard]] std::pair<SHAMapTreeNodePtr, SHAMapNodeID> const&
+        /**
+         * The node at the end of the path.
+         *
+         * @return a reference into the path, which a later push may invalidate
+         *         by reallocating, so a caller that pushes must ask again. An
+         *         empty path yields a null node the caller can test.
+         */
+        [[nodiscard]] SHAMapTreeNodePtr const&
         top() const
         {
-            XRPL_ASSERT(!stack_.empty(), "xrpl::SHAMap::NodePathStack::top : non-empty stack");
-            return stack_.top();
+            if (path_.empty())
+            {
+                // LCOV_EXCL_START
+                UNREACHABLE("xrpl::SHAMap::NodePathStack::top : empty stack");
+                static SHAMapTreeNodePtr const kEmpty;
+                return kEmpty;
+                // LCOV_EXCL_STOP
+            }
+            return path_.back();
         }
 
+        /**
+         * The depth of the node at the end of the path.
+         *
+         * @return the depth, which is the entry's own index. Zero on an empty
+         *         path.
+         */
+        [[nodiscard]] unsigned int
+        topDepth() const
+        {
+            if (path_.empty())
+            {
+                // LCOV_EXCL_START
+                UNREACHABLE("xrpl::SHAMap::NodePathStack::topDepth : empty stack");
+                return 0;
+                // LCOV_EXCL_STOP
+            }
+            return static_cast<unsigned int>(path_.size() - 1);
+        }
+
+        /**
+         * Shorten the path by one node. An empty path is left as it is.
+         */
         void
         pop()
         {
-            XRPL_ASSERT(!stack_.empty(), "xrpl::SHAMap::NodePathStack::pop : non-empty stack");
-            stack_.pop();
+            if (path_.empty())
+            {
+                // LCOV_EXCL_START
+                UNREACHABLE("xrpl::SHAMap::NodePathStack::pop : empty stack");
+                return;
+                // LCOV_EXCL_STOP
+            }
+            path_.pop_back();
         }
 
+        /**
+         * Discard the whole path.
+         */
         void
         clear()
         {
-            stack_ = {};
+            path_.clear();
+            pathKey_ = UInt256{};
         }
 
         /**
-         * Start a path at the root of the map, whose ID is the zero-depth ID by definition.
-         */
-        void
-        pushRoot(SHAMapTreeNodePtr node)
-        {
-            XRPL_ASSERT(stack_.empty(), "xrpl::SHAMap::NodePathStack::pushRoot : empty stack");
-            stack_.emplace(std::move(node), SHAMapNodeID{});
-        }
-
-        /**
-         * Extend the path to the child of the current node reached by `branch`.
+         * Shorten the path by one node and hand that node to the caller.
          *
-         * A node keeps the depth it was reached at, never a normalized kLeafDepth. Only a leaf may
-         * sit at kLeafDepth, since an inner node there would have no branch left to select.
-         */
-        void
-        pushChild(SHAMapTreeNodePtr node, unsigned int branch)
-        {
-            XRPL_ASSERT(node, "xrpl::SHAMap::NodePathStack::pushChild : non-null node input");
-            XRPL_ASSERT(
-                !stack_.empty(), "xrpl::SHAMap::NodePathStack::pushChild : non-empty stack");
-            auto childID = stack_.top().second.getChildNodeID(branch);
-            XRPL_ASSERT_IF(
-                node->isInner(),
-                childID.getDepth() < kLeafDepth,
-                "xrpl::SHAMap::NodePathStack::pushChild : inner node above leaf depth");
-            XRPL_ASSERT_IF(
-                node->isLeaf(),
-                childID.isPrefixOf(leafKey(*node)),
-                "xrpl::SHAMap::NodePathStack::pushChild : leaf key below branch");
-            stack_.emplace(std::move(node), std::move(childID));
-        }
-
-        /**
-         * Extend the path to a node lying on the path to `target`.
+         * Moves the node out, transferring the existing reference.
          *
-         * For nodes not reached by descending a known branch: the walk tracks only the key it is
-         * heading for, or the node is newly created. Either way `target` selects the branch.
+         * @return the node that was at the end of the path, or an empty
+         *         pointer if there was none.
          */
-        void
-        pushNode(SHAMapTreeNodePtr node, UInt256 const& target)
+        [[nodiscard]] SHAMapTreeNodePtr
+        releaseNode()
         {
-            if (stack_.empty())
+            if (path_.empty())
             {
-                pushRoot(std::move(node));
+                // LCOV_EXCL_START
+                UNREACHABLE("xrpl::SHAMap::NodePathStack::releaseNode : empty stack");
+                return {};
+                // LCOV_EXCL_STOP
+            }
+            auto node = std::move(path_.back());
+            path_.pop_back();
+            return node;
+        }
+
+        /**
+         * Shorten the path by one node and hand it over as a `Node`.
+         *
+         * @tparam Node SHAMapInnerNode or SHAMapLeafNode.
+         * @return the node that was at the end of the path, or an empty pointer
+         *         if the path was empty or that node is not a `Node`. The path
+         *         is shortened either way.
+         */
+        template <class Node>
+        [[nodiscard]] intr_ptr::SharedPtr<Node>
+        releaseNodeAs()
+        {
+            static_assert(
+                std::is_same_v<Node, SHAMapInnerNode> || std::is_same_v<Node, SHAMapLeafNode>,
+                "releaseNodeAs serves the two concrete node kinds");
+
+            auto node = releaseNode();
+            if (!node)
+            {
+                // The other half of releaseNode()'s empty-path contract, asserted there by the
+                // UNREACHABLE in its own LCOV_EXCL block.
+                return {};  // LCOV_EXCL_LINE
+            }
+
+            if constexpr (std::is_same_v<Node, SHAMapInnerNode>)
+            {
+                // A leaf sits only at the end of a path, which both callers asking for an inner
+                // node have released first. The other half of the contract their own UNREACHABLEs
+                // assert, in SHAMap::dirtyUp and SHAMap::delItem.
+                if (!node->isInner())
+                    return {};  // LCOV_EXCL_LINE
             }
             else
             {
-                pushChild(std::move(node), selectBranch(stack_.top().second, target));
+                if (!node->isLeaf())
+                    return {};
             }
+
+            return intr_ptr::staticPointerCast<Node>(std::move(node));
+        }
+
+        /**
+         * Start a path at the root of the map, which sits at depth zero by
+         * definition.
+         *
+         * @return false, leaving the path unchanged, if a path was already
+         *         started.
+         */
+        [[nodiscard]] bool
+        pushRoot(SHAMapTreeNodePtr node)
+        {
+            if (!path_.empty())
+            {
+                // LCOV_EXCL_START
+                UNREACHABLE("xrpl::SHAMap::NodePathStack::pushRoot : non-empty stack");
+                return false;
+                // LCOV_EXCL_STOP
+            }
+            // Reserved here rather than in the constructor, to keep SHAMap::end()
+            // allocation-free. A path holds at most kLeafDepth + 1 entries, so no push on a path
+            // started here reallocates. A copy is not given that room, so top()'s invalidation
+            // rule still holds.
+            path_.reserve(kLeafDepth + 1u);
+            path_.push_back(std::move(node));
+            return true;
+        }
+
+        /**
+         * Extend the path to the child of the node at its end reached by
+         * `branch`.
+         *
+         * Only a leaf may sit at kLeafDepth, since an inner node there would
+         * have no branch left to select.
+         *
+         * @param node the child to append.
+         * @param branch the branch of the current node that `node` was reached
+         *               through, which serves to judge the node offered.
+         * @return false if there is no node to descend from or to push, no
+         *         branch of that number, no room left below for the kind of
+         *         node offered, or a leaf whose own key does not select
+         *         `branch`. The path keeps its nodes.
+         */
+        [[nodiscard]] bool
+        pushChild(SHAMapTreeNodePtr node, unsigned int branch)
+        {
+            if (path_.empty() || !node || branch >= kBranchFactor)
+            {
+                // LCOV_EXCL_START
+                UNREACHABLE("xrpl::SHAMap::NodePathStack::pushChild : no child to push");
+                return false;
+                // LCOV_EXCL_STOP
+            }
+
+            // A node must have room below it at the depth it is offered for. Reachable, since a
+            // node resolved by hash carries no position of its own.
+            auto const parentDepth = topDepth();
+            bool const tooDeep = pastLeafDepth(parentDepth, *node);
+            SOMETIMES(tooDeep, "xrpl::SHAMap::NodePathStack::pushChild : child past leaf depth");
+            if (tooDeep)
+            {
+                return false;
+            }
+
+            // A leaf's key must agree with every branch recorded above it, which the chain in
+            // `pathKey_` supplies. A path's length gives only the depths.
+            pathKey_ = childNodeID(pathKey_, parentDepth, branch);
+
+            // The test takes the depth and the two keys, so no SHAMapNodeID is built: it
+            // is a CountedObject, and one per push would report as a live node ID.
+            bool const misplaced =
+                node->isLeaf() && !samePositionAtDepth(parentDepth + 1u, pathKey_, leafKey(*node));
+            SOMETIMES(
+                misplaced, "xrpl::SHAMap::NodePathStack::pushChild : leaf key outside branch");
+            if (misplaced)
+            {
+                return false;
+            }
+
+            path_.push_back(std::move(node));
+            return true;
+        }
+
+        /**
+         * Extend the path by one node lying on the way to `target`, starting
+         * it if it is empty.
+         *
+         * For nodes not reached by descending a known branch: the walk tracks
+         * only the key it is heading for, or the node is newly created. Either
+         * way `target` names the branch.
+         *
+         * @param node the node to append.
+         * @param target the key the walk is heading for.
+         * @return whatever pushRoot or pushChild returned.
+         */
+        [[nodiscard]] bool
+        pushNode(SHAMapTreeNodePtr node, UInt256 const& target)
+        {
+            if (path_.empty())
+            {
+                return pushRoot(std::move(node));
+            }
+            return pushChild(std::move(node), selectBranch(topDepth(), target));
         }
 
     private:
-        std::stack<std::pair<SHAMapTreeNodePtr, SHAMapNodeID>> stack_;
+        // path_[i] holds the node at depth i, by construction: pushRoot starts at depth 0 and
+        // pushChild only ever appends one level.
+        std::vector<SHAMapTreeNodePtr> path_;
+
+        // The branches descended, one nibble per level: nibble i is the branch taken from depth i.
+        // Nibbles below the path's end are stale after a pop, so every read masks at the path's
+        // own depth. childNodeID clears a nibble before writing it, so re-descending overwrites.
+        UInt256 pathKey_;
     };
 
     using DeltaRef =
         std::pair<boost::intrusive_ptr<SHAMapItem const>, boost::intrusive_ptr<SHAMapItem const>>;
 
+    /**
+     * The sequence of the ledger this map references, read atomically.
+     *
+     * @return The sequence, or zero if the map references no ledger.
+     */
+    [[nodiscard]] std::uint32_t
+    ledgerSeq() const;
+
+    /**
+     * The current state, read atomically.
+     *
+     * Orders state_ alone. The tree's nodes carry no ordering guarantees.
+     *
+     * @return The state as of the call, which a concurrent walk may already
+     *         have moved past.
+     */
+    [[nodiscard]] SHAMapState
+    state() const;
+
+    /**
+     * Record that the map is provably not the one it claims to be.
+     *
+     * Cannot fail, since Invalid outranks every other state. Const because a
+     * const descend() can reach this verdict while resolving a node.
+     */
+    void
+    setInvalid() const;
+
+    /**
+     * Condemn the map, naming in the log the node whose position proves it.
+     *
+     * Const, because a read path can reach this verdict. See setInvalid().
+     *
+     * @param node the node that cannot sit where it was offered.
+     * @param hash the hash it was resolved under.
+     * @param position the place in the tree it was offered for.
+     */
+    void
+    condemn(SHAMapTreeNode const& node, SHAMapHash const& hash, SHAMapNodeID const& position) const;
+
+    /**
+     * Move the map to a new state, atomically.
+     *
+     * With clearSynching(), the only writer of state_ past construction.
+     * Invalid is always stored, and every other transition is refused once
+     * the map is Invalid, which is what makes that verdict terminal.
+     * clearSynching() is narrower and moves only Synching.
+     *
+     * Const to let setInvalid() be const. state_ is mutable, which makes that
+     * legal.
+     *
+     * @param desired The state to move to.
+     * @return false if the map is Invalid and the requested state is not,
+     *         leaving it unchanged. True otherwise.
+     */
+    bool
+    trySetState(SHAMapState desired) const;
+
     // tree node cache operations
     SHAMapTreeNodePtr
     cacheLookup(SHAMapHash const& hash) const;
@@ -550,9 +919,13 @@ private:
     dirtyUp(NodePathStack& stack, UInt256 const& target, SHAMapTreeNodePtr terminal);
 
     /**
-     * Walk towards the specified id, returning the node.  Caller must check
-     *  if the return is nullptr, and if not, if the node->peekItem()->key() ==
-     * id
+     * Walk towards the specified id, returning the node.
+     *
+     * @param id the key to walk towards, which need not be in the map.
+     * @param stack records the path walked, or nullptr to skip recording it.
+     * @return the leaf the walk ended on, or nullptr if it ended on an inner
+     *         node or was refused. A returned leaf need not hold `id`, so
+     *         callers compare its key themselves.
      */
     SHAMapLeafNode*
     walkTowardsKey(UInt256 const& id, NodePathStack* stack = nullptr) const;
@@ -564,10 +937,16 @@ private:
 
     /**
      * Unshare the node, allowing it to be modified
+     *
+     * @param node the node to unshare.
+     * @param depth the depth the node sits at, which says whether it is the
+     *        root. A clone of the root has to be adopted as the new root. A
+     *        clone of any other node is hooked up by the caller.
+     * @return the node, cloned if it was shared.
      */
     template <class Node>
-    intr_ptr::SharedPtr<Node>
-    unshareNode(intr_ptr::SharedPtr<Node>, SHAMapNodeID const& nodeID);
+    [[nodiscard]] intr_ptr::SharedPtr<Node>
+    unshareNode(intr_ptr::SharedPtr<Node> node, unsigned int depth);
 
     /**
      * prepare a node to be modified before flushing
@@ -588,8 +967,13 @@ private:
     /**
      * Returns the first or last item at or below the node already on top of `stack`, extending
      * `stack` with the path walked to reach it.
+     *
+     * @param stack the path to extend, whose last node the search starts from.
+     * @param direction whether to take the lowest or the highest branch at
+     *                  each level.
+     * @return the leaf found, or nullptr if no leaf lies below that node.
      */
-    SHAMapLeafNode*
+    [[nodiscard]] SHAMapLeafNode*
     belowHelper(NodePathStack& stack, BelowDirection direction) const;
 
     /**
@@ -603,7 +987,7 @@ private:
      * @param direction First to search upwards from `id`, Last to search downwards.
      * @return An iterator to the item found, or end() if no item lies on that side of `id`.
      */
-    ConstIterator
+    [[nodiscard]] ConstIterator
     boundHelper(UInt256 const& id, BelowDirection direction) const;
 
     // Simple descent
@@ -628,6 +1012,22 @@ private:
         bool& pending,
         DescendCallback&&) const;
 
+    /**
+     * Resolve the child of `parent` on `branch`, judging where it lands.
+     *
+     * Const, yet it records the Invalid verdict through setInvalid(). A new
+     * caller must therefore be a path that may reach that verdict. Today
+     * addKnownNode() is the only one.
+     *
+     * @param parent the inner node to descend from.
+     * @param parentID the position of `parent`.
+     * @param branch the branch of `parent` to resolve, which the caller has
+     *               already bounded.
+     * @param filter an alternate source of nodes, or null for the store alone.
+     * @return the child and the child's position. A null child where the branch
+     *         could not be resolved, or where what came back does not belong at
+     *         that position, in which case the map is left Invalid.
+     */
     std::pair<SHAMapTreeNode*, SHAMapNodeID>
     descend(
         SHAMapInnerNode* parent,
@@ -728,10 +1128,29 @@ private:
     };
 
     // getMissingNodes helper functions
+
+    /**
+     * Examine the remaining branches of one inner node, recording or
+     * requesting what is missing.
+     *
+     * @param mn the walk's shared state, which collects the missing nodes.
+     * @param node the walk's current position, updated to the node to process
+     *             next.
+     */
     void
-    gmnProcessNodes(MissingNodes&, MissingNodes::StackEntry& node);
-    static void
-    gmnProcessDeferredReads(MissingNodes&);
+    gmnProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& node);
+
+    /**
+     * Wait for every read this pass posted, then hook up or record what each
+     * one resolved.
+     *
+     * Drains all of them even after judging the map, since an outstanding read
+     * holds a pointer to `mn`.
+     *
+     * @param mn the walk's shared state, holding the posted reads.
+     */
+    void
+    gmnProcessDeferredReads(MissingNodes& mn);
 
     // fetch from DB helper function
     SHAMapTreeNodePtr
@@ -741,44 +1160,116 @@ private:
 inline void
 SHAMap::setFull()
 {
-    full_ = true;
+    full_.store(true, std::memory_order_release);
 }
 
 inline void
 SHAMap::setLedgerSeq(std::uint32_t lseq)
 {
-    ledgerSeq_ = lseq;
+    ledgerSeq_.store(lseq, std::memory_order_relaxed);
 }
 
-inline void
+inline std::uint32_t
+SHAMap::ledgerSeq() const
+{
+    return ledgerSeq_.load(std::memory_order_relaxed);
+}
+
+inline SHAMapState
+SHAMap::state() const
+{
+    return state_.load(std::memory_order_acquire);
+}
+
+inline bool
+SHAMap::trySetState(SHAMapState desired) const
+{
+    // Invalid is stored outright: that verdict outranks a concurrent transition.
+    if (desired == SHAMapState::Invalid)
+    {
+        state_.store(SHAMapState::Invalid, std::memory_order_release);
+        return true;
+    }
+
+    // A failed exchange both reports the state and refreshes expected, so no load is needed
+    // ahead of the loop.
+    auto expected = SHAMapState::Modifying;
+    while (expected != SHAMapState::Invalid)
+    {
+        if (state_.compare_exchange_weak(
+                expected, desired, std::memory_order_acq_rel, std::memory_order_acquire))
+        {
+            return true;
+        }
+    }
+    return false;
+}
+
+inline bool
 SHAMap::setImmutable()
 {
-    XRPL_ASSERT(state_ != SHAMapState::Invalid, "xrpl::SHAMap::setImmutable : state is valid");
-    state_ = SHAMapState::Immutable;
+    SOMETIMES(!isValid(), "xrpl::SHAMap::setImmutable : map is invalid");
+    return trySetState(SHAMapState::Immutable);
 }
 
 inline bool
 SHAMap::isSynching() const
 {
-    return state_ == SHAMapState::Synching;
+    return state() == SHAMapState::Synching;
+}
+
+inline bool
+SHAMap::isImmutable() const
+{
+    return state() == SHAMapState::Immutable;
 }
 
 inline void
 SHAMap::setSynching()
 {
-    state_ = SHAMapState::Synching;
+    // Guarded, so this is not a way out of Invalid.
+    if (!trySetState(SHAMapState::Synching))
+    {
+        // Only ever called on a freshly constructed map.
+        // LCOV_EXCL_START
+        UNREACHABLE("xrpl::SHAMap::setSynching : map is invalid");
+        // LCOV_EXCL_STOP
+    }
 }
 
 inline void
 SHAMap::clearSynching()
 {
-    state_ = SHAMapState::Modifying;
+    // Only Synching moves. A walk can end after the ledger settled, and then the map stays
+    // Immutable; a Modifying map has nothing to clear; an invalid map stays invalid. Only that
+    // last refusal is logged, since a concurrent walk is what produces it.
+    auto expected = SHAMapState::Synching;
+    if (state_.compare_exchange_strong(
+            expected, SHAMapState::Modifying, std::memory_order_acq_rel, std::memory_order_acquire))
+    {
+        return;
+    }
+
+    bool const invalid = expected == SHAMapState::Invalid;
+    SOMETIMES(invalid, "xrpl::SHAMap::clearSynching : map is invalid");
+    if (invalid)
+    {
+        JLOG(journal_.warn()) << "Refused to clear synching on an invalid map, root hash "
+                              << root_->getHash();
+    }
 }
 
 inline bool
 SHAMap::isValid() const
 {
-    return state_ != SHAMapState::Invalid;
+    return state() != SHAMapState::Invalid;
+}
+
+inline void
+SHAMap::setInvalid() const
+{
+    // Through trySetState() like every other transition, so state_ has one writer funnel.
+    trySetState(SHAMapState::Invalid);
 }
 
 inline void

include/xrpl/shamap/SHAMapAddNode.h

@@ -12,47 +12,148 @@ private:
     int bad_;
     int duplicate_;
 
+    // Whether a node in the batch proved the map impossible. A fact about the batch rather than
+    // about the map, so a caller can tell this batch's doing from a concurrent walk's verdict.
+    bool invalidatedMap_;
+
 public:
     SHAMapAddNode();
+
+    /**
+     * Record one node that was rejected.
+     */
     void
     incInvalid();
+
+    /**
+     * Record one node that produced a good result.
+     */
     void
     incUseful();
+
+    /**
+     * Record one node that was not needed: the map already held it, or the map
+     * or its acquisition was no longer taking nodes. isGood() counts it on the
+     * accepted side.
+     */
     void
     incDuplicate();
-    void
-    reset();
+
+    /**
+     * @return How many nodes produced a good result.
+     */
     [[nodiscard]] int
     getGood() const;
+
+    /**
+     * @return How many nodes this code rejected, which is not how many the
+     *         batch was given: a batch that stops at its first bad node counts
+     *         one, and a batch that carries on counts each.
+     */
+    [[nodiscard]] int
+    getBad() const;
+
+    /**
+     * @return How many nodes were not needed, whether already held or offered
+     *         to a map no longer taking nodes, tallied apart from the good and
+     *         bad counts.
+     */
+    [[nodiscard]] int
+    getDuplicate() const;
+
+    /**
+     * Whether nodes that produced a good result or were not needed outnumber
+     * the ones rejected.
+     *
+     * @return Whether the batch was good.
+     */
     [[nodiscard]] bool
     isGood() const;
+
+    /**
+     * @return Whether at least one node in the batch was rejected.
+     */
     [[nodiscard]] bool
     isInvalid() const;
+
+    /**
+     * @return Whether at least one node in the batch produced a good result.
+     */
     [[nodiscard]] bool
     isUseful() const;
-    [[nodiscard]] std::string
-    get() const;
-
-    SHAMapAddNode&
-    operator+=(SHAMapAddNode const& n);
 
+    /**
+     * @return A verdict recording one node that was not needed.
+     */
     static SHAMapAddNode
     duplicate();
+
+    /**
+     * @return A verdict recording one useful node.
+     */
     static SHAMapAddNode
     useful();
+
+    /**
+     * @return A verdict recording one invalid node.
+     */
     static SHAMapAddNode
     invalid();
 
+    /**
+     * @return A verdict recording one invalid node that proved the map
+     *         impossible.
+     */
+    static SHAMapAddNode
+    mapInvalidated();
+
+    /**
+     * Whether a node in the batch proved the map impossible.
+     *
+     * Reports what the batch did rather than what the map now holds, so a
+     * verdict another thread's walk reached is not read as this batch's. The
+     * two differ: SHAMap::setInvalid() is written with no lock held, by a
+     * getMissingNodes() walk its caller runs lock-free.
+     *
+     * @return Whether a node in the batch invalidated the map.
+     */
+    [[nodiscard]] bool
+    invalidatedMap() const;
+
+    /**
+     * Clear every count back to zero.
+     */
+    void
+    reset();
+
+    /**
+     * Render the tally as a log line.
+     *
+     * @return The tally, e.g. "good:2 bad:1 dupe:1", or "no nodes processed" if
+     *         every count is zero.
+     */
+    [[nodiscard]] std::string
+    get() const;
+
+    /**
+     * Add another verdict's counts into this one.
+     *
+     * @param n The verdict to add.
+     * @return This verdict, updated.
+     */
+    SHAMapAddNode&
+    operator+=(SHAMapAddNode const& n);
+
 private:
-    SHAMapAddNode(int good, int bad, int duplicate);
+    SHAMapAddNode(int good, int bad, int duplicate, bool invalidatedMap = false);
 };
 
-inline SHAMapAddNode::SHAMapAddNode() : good_(0), bad_(0), duplicate_(0)
+inline SHAMapAddNode::SHAMapAddNode() : good_(0), bad_(0), duplicate_(0), invalidatedMap_(false)
 {
 }
 
-inline SHAMapAddNode::SHAMapAddNode(int good, int bad, int duplicate)
-    : good_(good), bad_(bad), duplicate_(duplicate)
+inline SHAMapAddNode::SHAMapAddNode(int good, int bad, int duplicate, bool invalidatedMap)
+    : good_(good), bad_(bad), duplicate_(duplicate), invalidatedMap_(invalidatedMap)
 {
 }
 
@@ -74,18 +175,30 @@ SHAMapAddNode::incDuplicate()
     ++duplicate_;
 }
 
-inline void
-SHAMapAddNode::reset()
-{
-    good_ = bad_ = duplicate_ = 0;
-}
-
 inline int
 SHAMapAddNode::getGood() const
 {
     return good_;
 }
 
+inline int
+SHAMapAddNode::getBad() const
+{
+    return bad_;
+}
+
+inline int
+SHAMapAddNode::getDuplicate() const
+{
+    return duplicate_;
+}
+
+inline bool
+SHAMapAddNode::isGood() const
+{
+    return (good_ + duplicate_) > bad_;
+}
+
 inline bool
 SHAMapAddNode::isInvalid() const
 {
@@ -98,20 +211,10 @@ SHAMapAddNode::isUseful() const
     return good_ > 0;
 }
 
-inline SHAMapAddNode&
-SHAMapAddNode::operator+=(SHAMapAddNode const& n)
-{
-    good_ += n.good_;
-    bad_ += n.bad_;
-    duplicate_ += n.duplicate_;
-
-    return *this;
-}
-
 inline bool
-SHAMapAddNode::isGood() const
+SHAMapAddNode::invalidatedMap() const
 {
-    return (good_ + duplicate_) > bad_;
+    return invalidatedMap_;
 }
 
 inline SHAMapAddNode
@@ -132,6 +235,19 @@ SHAMapAddNode::invalid()
     return SHAMapAddNode(0, 1, 0);
 }
 
+inline SHAMapAddNode
+SHAMapAddNode::mapInvalidated()
+{
+    return SHAMapAddNode(0, 1, 0, true);
+}
+
+inline void
+SHAMapAddNode::reset()
+{
+    good_ = bad_ = duplicate_ = 0;
+    invalidatedMap_ = false;
+}
+
 inline std::string
 SHAMapAddNode::get() const
 {
@@ -160,4 +276,17 @@ SHAMapAddNode::get() const
     return ret;
 }
 
+inline SHAMapAddNode&
+SHAMapAddNode::operator+=(SHAMapAddNode const& n)
+{
+    good_ += n.good_;
+    bad_ += n.bad_;
+    duplicate_ += n.duplicate_;
+
+    // Or-ed, not summed: one node in the batch having proved the map impossible is the whole fact.
+    invalidatedMap_ = invalidatedMap_ || n.invalidatedMap_;
+
+    return *this;
+}
+
 }  // namespace xrpl

include/xrpl/shamap/SHAMapInnerNode.h

@@ -19,7 +19,8 @@ class SHAMapInnerNode final : public SHAMapTreeNode, public CountedObject<SHAMap
 {
 public:
     /**
-     * Each inner node has 16 children (the 'radix tree' part of the map)
+     * Children per inner node: one branch per value of the key nibble that a
+     * node's depth selects.
      */
     static constexpr unsigned int kBranchFactor = 16;
 
@@ -31,7 +32,25 @@ private:
      */
     TaggedPointer hashesAndChildren_;
 
-    std::uint32_t fullBelowGen_ = 0;
+    // Pin that wrapping fullBelowGen_ in an atomic keeps the packed layout, and that the load
+    // isFullBelow() does once per node of every walk stays lock-free.
+    static_assert(std::atomic<std::uint32_t>::is_always_lock_free);
+    static_assert(sizeof(std::atomic<std::uint32_t>) == sizeof(std::uint32_t));
+    static_assert(alignof(std::atomic<std::uint32_t>) == alignof(std::uint32_t));
+
+    /**
+     * The generation in which a walk found every node below this one resident,
+     * with this node as the walk's root. The node object is shared by hash
+     * between maps and positions, and a hash commits to a node's contents
+     * rather than its place, so the flag is written and read for the root
+     * position only; below the root, the position-keyed FullBelowCache is the
+     * memo. Written from more than one thread, since canonicalization shares a
+     * node between maps and a walk can run with the acquisition lock released
+     * (see SHAMap::state_). Relaxed both ways, since a generation is only
+     * compared for equality and the children it vouches for are published
+     * through lock_.
+     */
+    std::atomic<std::uint32_t> fullBelowGen_ = 0;
     std::uint16_t isBranch_ = 0;
 
     /**
@@ -148,10 +167,23 @@ public:
     SHAMapTreeNodePtr
     canonicalizeChild(unsigned int branch, SHAMapTreeNodePtr node);
 
-    // sync functions
+    /**
+     * Whether a walk rooted at this node found every node below it resident.
+     *
+     * Meaningful only for the node a walk starts at; see fullBelowGen_.
+     *
+     * @param generation The FullBelowCache generation the asking walk runs in.
+     * @return True if a walk rooted here completed in that generation.
+     */
     bool
     isFullBelow(std::uint32_t generation) const;
 
+    /**
+     * Record that a walk rooted at this node found every node below it
+     * resident.
+     *
+     * @param gen The FullBelowCache generation the completing walk ran in.
+     */
     void
     setFullBelowGen(std::uint32_t gen);
 
@@ -204,13 +236,13 @@ SHAMapInnerNode::getBranchCount() const
 inline bool
 SHAMapInnerNode::isFullBelow(std::uint32_t generation) const
 {
-    return fullBelowGen_ == generation;
+    return fullBelowGen_.load(std::memory_order_relaxed) == generation;
 }
 
 inline void
 SHAMapInnerNode::setFullBelowGen(std::uint32_t gen)
 {
-    fullBelowGen_ = gen;
+    fullBelowGen_.store(gen, std::memory_order_relaxed);
 }
 
 }  // namespace xrpl

include/xrpl/shamap/SHAMapLeafNode.h

@@ -75,4 +75,41 @@ leafKey(SHAMapTreeNode const& node)
     return safeDowncast<SHAMapLeafNode const&>(node).peekItem()->key();
 }
 
+/**
+ * Whether a node may occupy a position in a SHAMap.
+ *
+ * A leaf's own key names its position. An inner node carries no key, so every
+ * position agrees with it and a caller's own depth rules are what bound it.
+ *
+ * @param nodeID the position the node is claimed to occupy.
+ * @param node the node to judge.
+ * @return whether the node's own key agrees with that position.
+ */
+[[nodiscard]] inline bool
+belongsAt(SHAMapNodeID const& nodeID, SHAMapTreeNode const& node)
+{
+    return !node.isLeaf() || nodeID.isPrefixOf(leafKey(node));
+}
+
+/**
+ * Whether a node may occupy the child position a branch leads to.
+ *
+ * The overload above, asked about a child without building the child's id.
+ *
+ * @param parentID the position of the node above, whose depth must be below
+ *                 SHAMap::kLeafDepth.
+ * @param branch the branch of that node leading to the position judged.
+ * @param node the node to judge.
+ * @return whether the node's own key agrees with that position.
+ */
+[[nodiscard]] inline bool
+belongsAt(SHAMapNodeID const& parentID, unsigned int branch, SHAMapTreeNode const& node)
+{
+    if (!node.isLeaf())
+        return true;
+
+    auto const& key = leafKey(node);
+    return parentID.isPrefixOf(key) && selectBranch(parentID, key) == branch;
+}
+
 }  // namespace xrpl

include/xrpl/shamap/SHAMapNodeID.h

@@ -123,6 +123,38 @@ operator<<(std::ostream& out, SHAMapNodeID const& node)
     return out << to_string(node);
 }
 
+/**
+ * Returns the id of the child a parent's branch leads to
+ *
+ * A child's id is its parent's with the nibble at the parent's own depth set to
+ * the branch taken. Every other nibble is copied through, so a masked parent
+ * gives a masked child. The write counterpart of selectBranch.
+ *
+ * @param parentID the id of the parent. The nibble at parentDepth is
+ *                 overwritten rather than merged, so an id already carrying one
+ *                 there is accepted.
+ * @param parentDepth the depth of the parent, below SHAMap::kLeafDepth.
+ * @param branch the branch of the parent leading to the child.
+ * @return that child's id, whose depth is parentDepth + 1.
+ */
+[[nodiscard]] UInt256
+childNodeID(UInt256 const& parentID, unsigned int parentDepth, unsigned int branch);
+
+/**
+ * Whether two keys name the same position at a given depth
+ *
+ * Both sides are masked, so either may carry bits below `depth`.
+ * SHAMapNodeID::isPrefixOf is the stricter form.
+ *
+ * @param depth how many leading nibbles name the position, at most
+ *              SHAMap::kLeafDepth. A greater depth is clamped to it.
+ * @param lhs one of the keys.
+ * @param rhs the other key.
+ * @return whether the two agree on those nibbles.
+ */
+[[nodiscard]] bool
+samePositionAtDepth(unsigned int depth, UInt256 const& lhs, UInt256 const& rhs);
+
 /**
  * Return an object representing a serialized SHAMap Node ID
  *
@@ -144,9 +176,26 @@ deserializeSHAMapNodeID(std::string_view s)
 /** @} */
 
 /**
- * Returns the branch that would contain the given hash
+ * Returns the branch at the given depth that would contain the given hash
+ *
+ * @param depth the depth of the node whose branch to select.
+ * @param hash the key whose nibble at that depth names the branch.
+ * @return the branch containing the hash.
  */
 [[nodiscard]] unsigned int
-selectBranch(SHAMapNodeID const& id, UInt256 const& hash);
+selectBranch(unsigned int depth, UInt256 const& hash);
+
+/**
+ * Returns the branch that would contain the given hash
+ *
+ * @param id the node whose depth to read.
+ * @param hash the key whose nibble at that depth names the branch.
+ * @return the branch containing the hash.
+ */
+[[nodiscard]] inline unsigned int
+selectBranch(SHAMapNodeID const& id, UInt256 const& hash)
+{
+    return selectBranch(id.getDepth(), hash);
+}
 
 }  // namespace xrpl

src/libxrpl/ledger/Ledger.cpp

@@ -202,7 +202,13 @@ Ledger::Ledger(
     }
 
     stateMap_.flushDirty(NodeObjectType::AccountNode);
-    setImmutable();
+    // Built locally. See Ledger::setImmutable().
+    if (!setImmutable())
+    {
+        // LCOV_EXCL_START
+        logicError("Ledger::Ledger(CreateGenesisT, ...): genesis ledger map is invalid");
+        // LCOV_EXCL_STOP
+    }
 }
 
 Ledger::Ledger(
@@ -213,7 +219,7 @@ Ledger::Ledger(
     Fees const& fees,
     Family& family,
     beast::Journal j)
-    : immutable_(true)
+    : immutable_(false)
     , txMap_(SHAMapType::TRANSACTION, info.txHash, family)
     , stateMap_(SHAMapType::STATE, info.accountHash, family)
     , fees_(fees)
@@ -236,8 +242,20 @@ Ledger::Ledger(
         JLOG(j.warn()) << "Don't have state data root for ledger" << header_.seq;
     }
 
-    txMap_.setImmutable();
-    stateMap_.setImmutable();
+    // Loaded locally. See Ledger::setImmutable().
+    if (setMapsImmutable())
+    {
+        immutable_ = true;
+    }
+    else
+    {
+        // LCOV_EXCL_START
+        JLOG(j.error()) << "Invalid map for ledger " << header_.seq;
+        UNREACHABLE("xrpl::Ledger::Ledger(LedgerHeader const&, ...) : map is invalid");
+        // Treat it as a damaged ledger: the code below recomputes the hash and re-acquires.
+        loaded = false;
+        // LCOV_EXCL_STOP
+    }
 
     if (!setup())
         loaded = false;
@@ -279,7 +297,8 @@ Ledger::Ledger(Ledger const& prevLedger, NetClock::time_point closeTime)
 }
 
 Ledger::Ledger(LedgerHeader const& info, Rules rules, Family& family)
-    : immutable_(true)
+    : immutable_(false)
+    , mapHashesFromHeader_(true)
     , txMap_(SHAMapType::TRANSACTION, info.txHash, family)
     , stateMap_(SHAMapType::STATE, info.accountHash, family)
     , rules_(std::move(rules))
@@ -308,27 +327,48 @@ Ledger::Ledger(
     setup();
 }
 
-void
+bool
 Ledger::setImmutable(bool rehash)
 {
-    // Force update, since this is the only
-    // place the hash transitions to valid
-    if (!immutable_ && rehash)
+    // An immutable ledger is treated as persistable, so an invalid map must never become
+    // immutable. The validity test runs before anything is written, so a refusal leaves
+    // the header as it was.
+    if (!mapsValid())
+        return false;
+
+    // Read now but written to the header below: getHash() can unshare a dirty tree, so it must run
+    // while the map is still mutable, while the header may only be written once both maps are
+    // settled. Skipped when the ledger is already immutable or when the header supplied these.
+    bool const deriveMapHashes = !immutable_ && !mapHashesFromHeader_ && rehash;
+    UInt256 const txHash = deriveMapHashes ? txMap_.getHash().asUInt256() : UInt256{};
+    UInt256 const accountHash = deriveMapHashes ? stateMap_.getHash().asUInt256() : UInt256{};
+
+    // A concurrent walk can invalidate a map between the check above and here (see
+    // SHAMap::state_), so the result is checked. That narrows the window rather than closing it,
+    // since setInvalid() outranks Immutable.
+    bool const bothImmutable = setMapsImmutable();
+    SOMETIMES(!bothImmutable, "xrpl::Ledger::setImmutable : map invalidated while going immutable");
+    if (!bothImmutable)
+        return false;  // LCOV_EXCL_LINE: only the walk named above reaches this, so no test does
+
+    // Written only now, so failing the check above leaves the header describing what the ledger was
+    // built from. Forced, since this is the only place the hash transitions to valid.
+    if (deriveMapHashes)
     {
-        header_.txHash = txMap_.getHash().asUInt256();
-        header_.accountHash = stateMap_.getHash().asUInt256();
+        header_.txHash = txHash;
+        header_.accountHash = accountHash;
     }
 
     if (rehash)
         header_.hash = calculateLedgerHash(header_);
 
+    // Set last, so isImmutable() reports only a ledger whose maps are both immutable.
     immutable_ = true;
-    txMap_.setImmutable();
-    stateMap_.setImmutable();
     setup();
+    return true;
 }
 
-void
+bool
 Ledger::setAccepted(
     NetClock::time_point closeTime,
     NetClock::duration closeResolution,
@@ -340,7 +380,17 @@ Ledger::setAccepted(
     header_.closeTime = closeTime;
     header_.closeTimeResolution = closeResolution;
     header_.closeFlags = correctCloseTime ? 0 : kSLcfNoConsensusTime;
-    setImmutable();
+    // Built locally. See Ledger::setImmutable().
+    if (!setImmutable())
+    {
+        // LCOV_EXCL_START
+        JLOG(j_.error()) << "Invalid map for accepted ledger " << header_.seq;
+        UNREACHABLE("xrpl::Ledger::setAccepted : map is invalid");
+        return false;
+        // LCOV_EXCL_STOP
+    }
+
+    return true;
 }
 
 bool

src/libxrpl/shamap/SHAMap.cpp

@@ -26,6 +26,7 @@
 
 #include <boost/smart_ptr/intrusive_ptr.hpp>
 
+#include <atomic>
 #include <cstdint>
 #include <exception>
 #include <functional>
@@ -77,14 +78,23 @@ SHAMap::SHAMap(SHAMap const& other, bool isMutable)
     : f_(other.f_)
     , journal_(other.f_.journal())
     , cowid_(other.cowid_ + 1)
-    , ledgerSeq_(other.ledgerSeq_)
+    , ledgerSeq_(other.ledgerSeq())
     , root_(other.root_)
-    , state_(isMutable ? SHAMapState::Modifying : SHAMapState::Immutable)
     , type_(other.type_)
     , backed_(other.backed_)
 {
+    // A snapshot shares the source's root, so Invalid carries over. Read once into a local, since
+    // the source's state can change while this runs.
+    auto const otherState = other.state();
+    auto const ownState = [&] {
+        if (otherState == SHAMapState::Invalid)
+            return SHAMapState::Invalid;
+        return isMutable ? SHAMapState::Modifying : SHAMapState::Immutable;
+    }();
+    state_.store(ownState, std::memory_order_release);
+
     // If either map may change, they cannot share nodes
-    if ((state_ != SHAMapState::Immutable) || (other.state_ != SHAMapState::Immutable))
+    if ((ownState != SHAMapState::Immutable) || (otherState != SHAMapState::Immutable))
     {
         unshare();
     }
@@ -96,6 +106,18 @@ SHAMap::snapShot(bool isMutable) const
     return std::make_shared<SHAMap>(*this, isMutable);
 }
 
+void
+SHAMap::condemn(SHAMapTreeNode const& node, SHAMapHash const& hash, SHAMapNodeID const& position)
+    const
+{
+    // journal_ is shared by every map built from one Family, so name which map this is. The root
+    // node's own hash, since getHash() unshares the tree on a zero hash.
+    JLOG(journal_.warn()) << (node.isLeaf() ? "Leaf " : "Inner node ") << hash << " cannot sit at "
+                          << position << ", so the map with root hash " << root_->getHash()
+                          << " is invalid";
+    setInvalid();
+}
+
 void
 SHAMap::dirtyUp(NodePathStack& stack, UInt256 const& target, SHAMapTreeNodePtr child)
 {
@@ -105,20 +127,25 @@ SHAMap::dirtyUp(NodePathStack& stack, UInt256 const& target, SHAMapTreeNodePtr c
     // child can be an inner node or a leaf
 
     XRPL_ASSERT(
-        (state_ != SHAMapState::Synching) && (state_ != SHAMapState::Immutable),
+        (state() != SHAMapState::Synching) && (state() != SHAMapState::Immutable),
         "xrpl::SHAMap::dirtyUp : valid state");
     XRPL_ASSERT(child && (child->cowid() == cowid_), "xrpl::SHAMap::dirtyUp : valid child input");
 
     while (!stack.empty())
     {
-        auto node = intr_ptr::dynamicPointerCast<SHAMapInnerNode>(stack.top().first);
-        SHAMapNodeID const nodeID = stack.top().second;
-        stack.pop();
-        XRPL_ASSERT(node, "xrpl::SHAMap::dirtyUp : non-null node");
+        auto const depth = stack.topDepth();
+        auto node = stack.releaseNodeAs<SHAMapInnerNode>();
+        if (!node)
+        {
+            // LCOV_EXCL_START
+            UNREACHABLE("xrpl::SHAMap::dirtyUp : node is not inner");
+            Throw<SHAMapMissingNode>(type_, target);
+            // LCOV_EXCL_STOP
+        }
 
-        auto const branch = selectBranch(nodeID, target);
+        auto const branch = selectBranch(depth, target);
 
-        node = unshareNode(std::move(node), nodeID);
+        node = unshareNode(std::move(node), depth);
         node->setChild(branch, std::move(child));
 
         child = std::move(node);
@@ -128,32 +155,62 @@ SHAMap::dirtyUp(NodePathStack& stack, UInt256 const& target, SHAMapTreeNodePtr c
 SHAMapLeafNode*
 SHAMap::walkTowardsKey(UInt256 const& id, NodePathStack* stack) const
 {
-    XRPL_ASSERT(
-        stack == nullptr || stack->empty(), "xrpl::SHAMap::walkTowardsKey : empty stack input");
-    auto inNode = root_;
-    SHAMapNodeID nodeID;
+    if (stack != nullptr && !stack->empty())
+    {
+        // The path must start at the root, so a non-empty one is cleared rather than appended to.
+        // LCOV_EXCL_START
+        UNREACHABLE("xrpl::SHAMap::walkTowardsKey : non-empty stack input");
+        stack->clear();
+        return nullptr;
+        // LCOV_EXCL_STOP
+    }
 
-    // Every node on this walk lies on the path to `id`, so the stack can derive each ID from the
-    // branch `id` selects at the node above it.
-    auto pushCurrent = [&] {
-        if (stack != nullptr)
-            stack->pushNode(inNode, id);
+    auto inNode = root_;
+    unsigned int noStackDepth = 0;
+
+    // Returns false when the path could not be extended, which is separate from `id` being absent
+    // (see pushChild). The path is cleared, so a caller reads an empty path as a refusal.
+    auto pushCurrent = [&]() -> bool {
+        if (stack == nullptr || stack->pushNode(inNode, id))
+        {
+            return true;
+        }
+        stack->clear();
+        return false;
     };
 
     while (inNode->isInner())
     {
-        pushCurrent();
+        if (!pushCurrent())
+        {
+            return nullptr;
+        }
 
         auto& inner = safeDowncast<SHAMapInnerNode&>(*inNode);
-        auto const branch = selectBranch(nodeID, id);
+        auto const depth = stack != nullptr ? stack->topDepth() : noStackDepth;
+        auto const branch = selectBranch(depth, id);
         if (inner.isEmptyBranch(branch))
             return nullptr;
 
         inNode = descendThrow(inner, branch);
-        nodeID = nodeID.getChildNodeID(branch);
+        if (stack == nullptr)
+        {
+            // The same depth bound pushChild applies, so both modes refuse at the same depth. A
+            // node resolved from the local store still has its position and type to be judged.
+            bool const tooDeep = pastLeafDepth(depth, *inNode);
+            SOMETIMES(tooDeep, "xrpl::SHAMap::walkTowardsKey : child too deep");
+            if (tooDeep)
+            {
+                return nullptr;
+            }
+            ++noStackDepth;
+        }
     }
 
-    pushCurrent();
+    if (!pushCurrent())
+    {
+        return nullptr;
+    }
     return safeDowncast<SHAMapLeafNode*>(inNode.get());
 }
 
@@ -170,7 +227,7 @@ SHAMapTreeNodePtr
 SHAMap::fetchNodeFromDB(SHAMapHash const& hash) const
 {
     XRPL_ASSERT(backed_, "xrpl::SHAMap::fetchNodeFromDB : is backed");
-    auto obj = f_.db().fetchNodeObject(hash.asUInt256(), ledgerSeq_);
+    auto obj = f_.db().fetchNodeObject(hash.asUInt256(), ledgerSeq());
     return finishFetch(hash, obj);
 }
 
@@ -183,10 +240,14 @@ SHAMap::finishFetch(SHAMapHash const& hash, std::shared_ptr<NodeObject> const& o
     {
         if (!object)
         {
-            if (full_)
+            // A missing node disproves full_, so withdraw it and report the gap. The acq_rel
+            // exchange orders the report and makes exactly one reader report it. The load ahead
+            // of it is relaxed, since all it decides is whether to try the exchange: a stale true
+            // costs one exchange that loses, and a stale false means another reader won already.
+            if (full_.load(std::memory_order_relaxed) &&
+                full_.exchange(false, std::memory_order_acq_rel))
             {
-                full_ = false;
-                f_.missingNodeAcquireBySeq(ledgerSeq_, hash.asUInt256());
+                f_.missingNodeAcquireBySeq(ledgerSeq(), hash.asUInt256());
             }
             return {};
         }
@@ -219,7 +280,7 @@ SHAMap::checkFilter(SHAMapHash const& hash, SHAMapSyncFilter const* filter) cons
             auto node = SHAMapTreeNode::makeFromPrefix(makeSlice(*nodeData), hash);
             if (node)
             {
-                filter->gotNode(true, hash, ledgerSeq_, std::move(*nodeData), node->getType());
+                filter->gotNode(true, hash, ledgerSeq(), std::move(*nodeData), node->getType());
                 if (backed_)
                     canonicalize(hash, node);
             }
@@ -357,12 +418,23 @@ SHAMap::descend(
         !parent->isEmptyBranch(branch), "xrpl::SHAMap::descend : parent branch is non-empty");
 
     SHAMapTreeNode* child = parent->getChildPointer(branch);  // NOLINT(misc-const-correctness)
+    auto childID = parentID.getChildNodeID(branch);
 
     if (child == nullptr)
     {
         auto const& childHash = parent->getChildHash(branch);
         SHAMapTreeNodePtr childNode = fetchNodeNT(childHash, filter);
 
+        if (childNode &&
+            (!belongsAt(childID, *childNode) || pastLeafDepth(parentID.getDepth(), *childNode)))
+        {
+            // A hash covers a node's contents, not its position, so position is judged before
+            // canonicalizeChild hooks the node in. The verdict lands on the map, since resolving
+            // the same blob again gives the same node.
+            condemn(*childNode, childHash, childID);
+            return std::make_pair(nullptr, std::move(childID));
+        }
+
         if (childNode)
         {
             childNode = parent->canonicalizeChild(branch, std::move(childNode));
@@ -370,7 +442,7 @@ SHAMap::descend(
         }
     }
 
-    return std::make_pair(child, parentID.getChildNodeID(branch));
+    return std::make_pair(child, std::move(childID));
 }
 
 SHAMapTreeNode*
@@ -399,7 +471,7 @@ SHAMap::descendAsync(
         {
             f_.db().asyncFetch(
                 hash.asUInt256(),
-                ledgerSeq_,
+                ledgerSeq(),
                 [this, hash, cb{std::move(callback)}](std::shared_ptr<NodeObject> const& object) {
                     auto node = finishFetch(hash, object);
                     cb(node, hash);
@@ -417,16 +489,16 @@ SHAMap::descendAsync(
 
 template <class Node>
 intr_ptr::SharedPtr<Node>
-SHAMap::unshareNode(intr_ptr::SharedPtr<Node> node, SHAMapNodeID const& nodeID)
+SHAMap::unshareNode(intr_ptr::SharedPtr<Node> node, unsigned int depth)
 {
     // make sure the node is suitable for the intended operation (copy on write)
     XRPL_ASSERT(node->cowid() <= cowid_, "xrpl::SHAMap::unshareNode : node valid for cowid");
     if (node->cowid() != cowid_)
     {
         // have a CoW
-        XRPL_ASSERT(state_ != SHAMapState::Immutable, "xrpl::SHAMap::unshareNode : not immutable");
+        XRPL_ASSERT(state() != SHAMapState::Immutable, "xrpl::SHAMap::unshareNode : not immutable");
         node = intr_ptr::staticPointerCast<Node>(node->clone(cowid_));
-        if (nodeID.isRoot())
+        if (depth == 0)
             root_ = node;
     }
     return node;
@@ -436,14 +508,19 @@ SHAMapLeafNode*
 SHAMap::belowHelper(NodePathStack& stack, BelowDirection direction) const
 {
     XRPL_ASSERT(!stack.empty(), "xrpl::SHAMap::belowHelper : non-empty stack input");
-    if (auto const& top = stack.top().first; top->isLeaf())
+    if (stack.empty())
+    {
+        // LCOV_EXCL_START
+        return nullptr;
+        // LCOV_EXCL_STOP
+    }
+    if (auto const& top = stack.top(); top->isLeaf())
         return safeDowncast<SHAMapLeafNode*>(top.get());
 
-    // The stack owns the node/ID pairing, so descending is only ever "push the branch we took".
-    // `scanned` counts how many branches of the current node we have examined; the branch we look
-    // at is derived from it, so no index ever goes out of range. `inner` tracks the node on top of
-    // the stack, which keeps it alive, so it only needs recomputing after a push.
-    auto* inner = safeDowncast<SHAMapInnerNode*>(stack.top().first.get());
+    // `scanned` counts the branches of the current node already examined, and the branch looked at
+    // is derived from it, so the index stays inside the fan-out. `inner` points at the top of the
+    // path, which keeps it alive, so it only needs recomputing after a push.
+    auto* inner = safeDowncast<SHAMapInnerNode*>(stack.top().get());
     for (auto scanned = 0u; scanned < kBranchFactor;)
     {
         auto const childBranch =
@@ -455,9 +532,20 @@ SHAMap::belowHelper(NodePathStack& stack, BelowDirection direction) const
             continue;
         }
 
-        stack.pushChild(descendThrow(*inner, childBranch), childBranch);
+        auto const parentDepth = stack.topDepth();
+        auto descended = descendThrow(*inner, childBranch);
+        if (!stack.pushChild(std::move(descended), childBranch))
+        {
+            // Throwing rather than returning nullptr keeps nullptr meaning a subtree with no leaf
+            // below it, which begin() and an iterator increment both rely on. Every caller here is
+            // a const read on a shared snapshot, so the verdict is left to the acquisition path
+            // (see SHAMap::descend and gmnProcessNodes).
+            JLOG(journal_.warn()) << "Cannot walk below depth " << parentDepth << " at branch "
+                                  << childBranch;
+            Throw<SHAMapMissingNode>(type_, inner->getChildHash(childBranch));
+        }
 
-        auto const& child = stack.top().first;
+        auto const& child = stack.top();
         if (child->isLeaf())
             return safeDowncast<SHAMapLeafNode*>(child.get());
 
@@ -512,7 +600,12 @@ SHAMapLeafNode const*
 SHAMap::peekFirstItem(NodePathStack& stack) const
 {
     XRPL_ASSERT(stack.empty(), "xrpl::SHAMap::peekFirstItem : empty stack input");
-    stack.pushRoot(root_);
+    if (!stack.pushRoot(root_))
+    {
+        // LCOV_EXCL_START
+        return nullptr;
+        // LCOV_EXCL_STOP
+    }
     SHAMapLeafNode const* node = belowHelper(stack, BelowDirection::First);
     if (node == nullptr)
     {
@@ -526,18 +619,28 @@ SHAMapLeafNode const*
 SHAMap::peekNextItem(UInt256 const& id, NodePathStack& stack) const
 {
     XRPL_ASSERT(!stack.empty(), "xrpl::SHAMap::peekNextItem : non-empty stack input");
-    XRPL_ASSERT(stack.top().first->isLeaf(), "xrpl::SHAMap::peekNextItem : stack starts with leaf");
+    if (stack.empty())
+    {
+        // LCOV_EXCL_START
+        return nullptr;
+        // LCOV_EXCL_STOP
+    }
+    XRPL_ASSERT(stack.top()->isLeaf(), "xrpl::SHAMap::peekNextItem : stack starts with leaf");
     stack.pop();
     while (!stack.empty())
     {
-        auto const [node, nodeID] = stack.top();
+        auto const& node = stack.top();
         XRPL_ASSERT(!node->isLeaf(), "xrpl::SHAMap::peekNextItem : another node is not leaf");
         auto& inner = safeDowncast<SHAMapInnerNode&>(*node);
-        for (auto i = selectBranch(nodeID, id) + 1; i < kBranchFactor; ++i)
+        for (auto i = selectBranch(stack.topDepth(), id) + 1; i < kBranchFactor; ++i)
         {
             if (!inner.isEmptyBranch(i))
             {
-                stack.pushChild(descendThrow(inner, i), i);
+                auto child = descendThrow(inner, i);
+                if (!stack.pushChild(std::move(child), i))
+                {
+                    Throw<SHAMapMissingNode>(type_, id);
+                }
                 auto leaf = belowHelper(stack, BelowDirection::First);
                 if (leaf == nullptr)
                     Throw<SHAMapMissingNode>(type_, id);
@@ -581,9 +684,15 @@ SHAMap::boundHelper(UInt256 const& id, BelowDirection direction) const
 
     NodePathStack stack;
     walkTowardsKey(id, &stack);
+
+    // An empty path means the walk refused a node. An empty map still leaves its root on the path,
+    // and end() would claim no key lies on the requested side of `id`, so a refusal throws.
+    if (stack.empty())
+        Throw<SHAMapMissingNode>(type_, id);
+
     while (!stack.empty())
     {
-        auto const [node, nodeID] = stack.top();
+        auto const& node = stack.top();
         if (node->isLeaf())
         {
             auto const& item = safeDowncast<SHAMapLeafNode const&>(*node).peekItem();
@@ -593,7 +702,7 @@ SHAMap::boundHelper(UInt256 const& id, BelowDirection direction) const
         else
         {
             auto& inner = safeDowncast<SHAMapInnerNode&>(*node);
-            auto const taken = selectBranch(nodeID, id);
+            auto const taken = selectBranch(stack.topDepth(), id);
             auto const remaining = searchingForward ? (kBranchFactor - 1u - taken) : taken;
 
             for (auto scanned = 0u; scanned < remaining; ++scanned)
@@ -603,7 +712,11 @@ SHAMap::boundHelper(UInt256 const& id, BelowDirection direction) const
                 if (inner.isEmptyBranch(branch))
                     continue;
 
-                stack.pushChild(descendThrow(inner, branch), branch);
+                auto child = descendThrow(inner, branch);
+                if (!stack.pushChild(std::move(child), branch))
+                {
+                    Throw<SHAMapMissingNode>(type_, id);
+                }
                 auto const leaf = belowHelper(stack, direction);
                 if (leaf == nullptr)
                     Throw<SHAMapMissingNode>(type_, id);
@@ -637,7 +750,7 @@ bool
 SHAMap::delItem(UInt256 const& id)
 {
     // delete the item with this ID
-    XRPL_ASSERT(state_ != SHAMapState::Immutable, "xrpl::SHAMap::delItem : not immutable");
+    XRPL_ASSERT(state() != SHAMapState::Immutable, "xrpl::SHAMap::delItem : not immutable");
 
     NodePathStack stack;
     walkTowardsKey(id, &stack);
@@ -645,10 +758,12 @@ SHAMap::delItem(UInt256 const& id)
     if (stack.empty())
         Throw<SHAMapMissingNode>(type_, id);
 
-    auto leaf = intr_ptr::dynamicPointerCast<SHAMapLeafNode>(stack.top().first);
-    stack.pop();
+    // An absent id leaves an inner node on top, which is the "not found" answer.
+    auto leaf = stack.releaseNodeAs<SHAMapLeafNode>();
+    if (!leaf)
+        return false;
 
-    if (!leaf || (leaf->peekItem()->key() != id))
+    if (leaf->peekItem()->key() != id)
         return false;
 
     SHAMapNodeType const type = leaf->getType();
@@ -658,19 +773,25 @@ SHAMap::delItem(UInt256 const& id)
 
     while (!stack.empty())
     {
-        auto node = intr_ptr::staticPointerCast<SHAMapInnerNode>(stack.top().first);
-        SHAMapNodeID const nodeID = stack.top().second;
-        stack.pop();
+        auto const depth = stack.topDepth();
+        auto node = stack.releaseNodeAs<SHAMapInnerNode>();
+        if (!node)
+        {
+            // LCOV_EXCL_START
+            UNREACHABLE("xrpl::SHAMap::delItem : node is not inner");
+            Throw<SHAMapMissingNode>(type_, id);
+            // LCOV_EXCL_STOP
+        }
 
-        node = unshareNode(std::move(node), nodeID);
+        node = unshareNode(std::move(node), depth);
         node->setChild(
-            selectBranch(nodeID, id), std::move(prevNode));  // NOLINT(bugprone-use-after-move)
+            selectBranch(depth, id), std::move(prevNode));  // NOLINT(bugprone-use-after-move)
 
         XRPL_ASSERT(
             not prevNode,  // NOLINT(bugprone-use-after-move)
             "xrpl::SHAMap::delItem : prevNode should be nullptr after std::move");
 
-        if (!nodeID.isRoot())
+        if (depth != 0)
         {
             // we may have made this a node with 1 or 0 children
             // And, if so, we need to remove this branch
@@ -719,7 +840,7 @@ SHAMap::delItem(UInt256 const& id)
 bool
 SHAMap::addGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem const> item)
 {
-    XRPL_ASSERT(state_ != SHAMapState::Immutable, "xrpl::SHAMap::addGiveItem : not immutable");
+    XRPL_ASSERT(state() != SHAMapState::Immutable, "xrpl::SHAMap::addGiveItem : not immutable");
     XRPL_ASSERT(type != SHAMapNodeType::TnInner, "xrpl::SHAMap::addGiveItem : valid type input");
 
     // add the specified item, does not update
@@ -731,8 +852,8 @@ SHAMap::addGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem const>
     if (stack.empty())
         Throw<SHAMapMissingNode>(type_, tag);
 
-    auto [node, nodeID] = stack.top();
-    stack.pop();
+    auto depth = stack.topDepth();
+    auto node = stack.releaseNode();
 
     if (node->isLeaf())
     {
@@ -740,12 +861,12 @@ SHAMap::addGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem const>
         if (leaf->peekItem()->key() == tag)
             return false;
     }
-    node = unshareNode(std::move(node), nodeID);
+    node = unshareNode(std::move(node), depth);
     if (node->isInner())
     {
         // easy case, we end on an inner node
         auto inner = intr_ptr::staticPointerCast<SHAMapInnerNode>(node);
-        auto const branch = selectBranch(nodeID, tag);
+        auto const branch = selectBranch(depth, tag);
         XRPL_ASSERT(
             inner->isEmptyBranch(branch), "xrpl::SHAMap::addGiveItem : inner branch is empty");
         inner->setChild(branch, makeTypedLeaf(type, std::move(item), cowid_));
@@ -763,13 +884,20 @@ SHAMap::addGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem const>
 
         auto b1 = 0u, b2 = 0u;
 
-        while ((b1 = selectBranch(nodeID, tag)) == (b2 = selectBranch(nodeID, otherItem->key())))
+        while ((b1 = selectBranch(depth, tag)) == (b2 = selectBranch(depth, otherItem->key())))
         {
-            stack.pushNode(node, tag);
+            if (!stack.pushNode(node, tag))
+            {
+                // The loop advances only while the two keys agree at the current nibble, and keys
+                // agreeing at every nibble are equal, which returned false above.
+                // LCOV_EXCL_START
+                Throw<SHAMapMissingNode>(type_, tag);
+                // LCOV_EXCL_STOP
+            }
 
             // we need a new inner node, since both go on same branch at this
             // level
-            nodeID = nodeID.getChildNodeID(b1);
+            ++depth;
             node = intr_ptr::makeShared<SHAMapInnerNode>(cowid_);
         }
 
@@ -810,7 +938,7 @@ SHAMap::updateGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem cons
     // can't change the tag but can change the hash
     UInt256 const tag = item->key();
 
-    XRPL_ASSERT(state_ != SHAMapState::Immutable, "xrpl::SHAMap::updateGiveItem : not immutable");
+    XRPL_ASSERT(state() != SHAMapState::Immutable, "xrpl::SHAMap::updateGiveItem : not immutable");
 
     NodePathStack stack;
     walkTowardsKey(tag, &stack);
@@ -818,16 +946,19 @@ SHAMap::updateGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem cons
     if (stack.empty())
         Throw<SHAMapMissingNode>(type_, tag);
 
-    auto node = intr_ptr::dynamicPointerCast<SHAMapLeafNode>(stack.top().first);
-    auto nodeID = stack.top().second;
-    stack.pop();
+    auto const depth = stack.topDepth();
 
-    if (!node || (node->peekItem()->key() != tag))
+    // A tag absent from the map leaves an inner node on top, which the API permits.
+    auto node = stack.releaseNodeAs<SHAMapLeafNode>();
+    if (!node)
+    {
+        return false;
+    }
+
+    // Or the walk ended on a leaf holding some other key.
+    if (node->peekItem()->key() != tag)
     {
-        // LCOV_EXCL_START
-        UNREACHABLE("xrpl::SHAMap::updateGiveItem : invalid node");
         return false;
-        // LCOV_EXCL_STOP
     }
 
     if (node->getType() != type)
@@ -836,7 +967,7 @@ SHAMap::updateGiveItem(SHAMapNodeType type, boost::intrusive_ptr<SHAMapItem cons
         return false;
     }
 
-    node = unshareNode(std::move(node), nodeID);
+    node = unshareNode(std::move(node), depth);
 
     if (node->setItem(item))
         dirtyUp(stack, tag, node);
@@ -901,7 +1032,7 @@ SHAMap::writeNode(NodeObjectType t, SHAMapTreeNodePtr node) const
 
     Serializer s;
     node->serializeWithPrefix(s);
-    f_.db().store(t, std::move(s.modData()), node->getHash().asUInt256(), ledgerSeq_);
+    f_.db().store(t, std::move(s.modData()), node->getHash().asUInt256(), ledgerSeq());
     return node;
 }
 

src/libxrpl/shamap/SHAMapInnerNode.cpp

@@ -16,6 +16,7 @@
 #include <xrpl/shamap/detail/TaggedPointer.h>
 #include <xrpl/shamap/detail/TaggedPointer.ipp>
 
+#include <atomic>
 #include <cstddef>
 #include <cstdint>
 #include <mutex>
@@ -77,7 +78,9 @@ SHAMapInnerNode::clone(std::uint32_t cowid) const
     auto p = intr_ptr::makeShared<SHAMapInnerNode>(cowid, branchCount);
     p->hash_ = hash_;
     p->isBranch_ = isBranch_;
-    p->fullBelowGen_ = fullBelowGen_;
+    // Relaxed, as everywhere. p is not reachable by another thread until this returns.
+    p->fullBelowGen_.store(
+        fullBelowGen_.load(std::memory_order_relaxed), std::memory_order_relaxed);
     SHAMapHash* cloneHashes = nullptr;
     SHAMapHash* thisHashes = nullptr;
     SHAMapTreeNodePtr* cloneChildren = nullptr;

src/libxrpl/shamap/SHAMapNodeID.cpp

@@ -50,14 +50,31 @@ maskedToDepth(UInt256 const& key, unsigned int depth)
     return key & depthMask(depth);
 }
 
-// Whether `id` at `depth` is what `key` looks like once masked down to that depth, i.e.
-// whether an ID with this depth and id names a subtree that `key` falls under.
+// Whether `key` masked to `depth` equals `id`. Asymmetric: `id` is not masked, so this also
+// refuses an id carrying bits below its own depth.
 static bool
 isPrefixOfAtDepth(UInt256 const& id, unsigned int depth, UInt256 const& key)
 {
     return maskedToDepth(key, depth) == id;
 }
 
+bool
+samePositionAtDepth(unsigned int depth, UInt256 const& lhs, UInt256 const& rhs)
+{
+    // The mask is chosen here, so an out-of-range depth would index depthMask's table past its
+    // last entry. kLeafDepth itself is in range, unlike in selectBranch: the position at the leaf
+    // depth is the whole key. A public helper has to hold its own bound.
+    if (depth > SHAMap::kLeafDepth)
+    {
+        // LCOV_EXCL_START
+        UNREACHABLE("xrpl::samePositionAtDepth : depth within tree");
+        depth = SHAMap::kLeafDepth;
+        // LCOV_EXCL_STOP
+    }
+
+    return maskedToDepth(lhs, depth) == maskedToDepth(rhs, depth);
+}
+
 // canonicalize the hash to a node ID for this depth
 SHAMapNodeID::SHAMapNodeID(unsigned int depth, UInt256 const& hash) : id_(hash), depth_(depth)
 {
@@ -90,6 +107,31 @@ SHAMapNodeID::getRawString() const
     return s.getString();
 }
 
+[[nodiscard]] UInt256
+childNodeID(UInt256 const& parentID, unsigned int parentDepth, unsigned int branch)
+{
+    XRPL_ASSERT(branch < SHAMap::kBranchFactor, "xrpl::childNodeID : valid branch input");
+    XRPL_ASSERT(parentDepth < SHAMap::kLeafDepth, "xrpl::childNodeID : parent above leaf depth");
+
+    // Only a depth below kLeafDepth names a child nibble, so clamp to the last one that does.
+    auto const clamped = std::min(parentDepth, SHAMap::kLeafDepth - 1u);
+
+    // The nibble is cleared before the branch is written, so an id already carrying one there is
+    // overwritten rather than merged. The byte index and both nibble choices read `clamped`, so
+    // they cannot disagree.
+    UInt256 id = parentID;
+    auto& byte = *(id.begin() + (clamped / 2));
+    if ((clamped & 1) != 0u)
+    {
+        byte = static_cast<unsigned char>((byte & 0xF0u) | branch);
+    }
+    else
+    {
+        byte = static_cast<unsigned char>((byte & 0x0Fu) | (branch << 4));
+    }
+    return id;
+}
+
 SHAMapNodeID
 SHAMapNodeID::getChildNodeID(unsigned int branch) const
 {
@@ -107,15 +149,13 @@ SHAMapNodeID::getChildNodeID(unsigned int branch) const
     XRPL_ASSERT(
         depth_ <= SHAMap::kLeafDepth, "xrpl::SHAMapNodeID::getChildNodeID : maximum leaf depth");
 
-    if (depth_ >= SHAMap::kLeafDepth)
+    if (SHAMap::isLeafDepth(depth_))
         Throw<std::logic_error>(std::format("Request for child node ID of {}", to_string(*this)));
 
     if (!isPrefixOf(id_))
         Throw<std::logic_error>(std::format("Incorrect mask for {}", to_string(*this)));
 
-    SHAMapNodeID node{depth_ + 1, id_};
-    node.id_.begin()[depth_ / 2] |= ((depth_ & 1) != 0u) ? branch : (branch << 4);
-    return node;
+    return SHAMapNodeID{depth_ + 1, childNodeID(id_, depth_, branch)};
 }
 
 bool
@@ -145,16 +185,16 @@ deserializeSHAMapNodeID(void const* data, std::size_t size)
 }
 
 [[nodiscard]] unsigned int
-selectBranch(SHAMapNodeID const& id, UInt256 const& hash)
+selectBranch(unsigned int depth, UInt256 const& hash)
 {
-    XRPL_ASSERT(id.getDepth() < SHAMap::kLeafDepth, "xrpl::selectBranch : depth below leaf depth");
+    XRPL_ASSERT(depth < SHAMap::kLeafDepth, "xrpl::selectBranch : depth below leaf depth");
 
-    // A depth-64 ID has no nibble left to select. Callers must not ask, but clamp anyway to keep
-    // the read below the end of the 32-byte key.
-    auto const depth = std::min(id.getDepth(), SHAMap::kLeafDepth - 1u);
-    auto branch = static_cast<unsigned int>(*(hash.begin() + (depth / 2)));
+    // Only a depth below kLeafDepth has a nibble to select, so clamp to the last one that does.
+    auto const clamped = std::min(depth, SHAMap::kLeafDepth - 1u);
+    auto branch = static_cast<unsigned int>(*(hash.begin() + (clamped / 2)));
 
-    if ((depth & 1) != 0u)
+    // The byte index and the nibble choice both read `clamped`, so they cannot disagree.
+    if ((clamped & 1) != 0u)
     {
         branch &= 0xf;
     }

src/libxrpl/shamap/SHAMapSync.cpp

@@ -143,13 +143,10 @@ SHAMap::visitDifferences(
         if (!function(*node))
             return;
 
-        // Nibbles run out at kLeafDepth, so only a leaf belongs there. A well-formed map never
-        // holds an inner node at that depth: addKnownNode marks the map invalid rather than hooking
-        // one in, and fetch-pack data is hash-verified against a validated root, so reaching this
-        // means a defect or a corrupt store, not something a peer can provoke. Report the node
-        // anyway - the wire form carries no depth, and the recipient hooks blobs in by hash - but
-        // skip the children rather than letting getChildNodeID throw on them.
-        if (nodeID.getDepth() >= kLeafDepth)
+        // Only a leaf belongs at kLeafDepth. The node is still reported, since the wire form
+        // carries no depth and the recipient hooks blobs by hash. Its children are skipped, since
+        // getChildNodeID has no answer past kLeafDepth.
+        if (isLeafDepth(nodeID.getDepth()))
         {
             // LCOV_EXCL_START
             UNREACHABLE("xrpl::SHAMap::visitDifferences : inner node at leaf depth");
@@ -207,7 +204,12 @@ SHAMap::gmnProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
             // we already know this child node is missing
             fullBelow = false;
         }
-        else if (!backed_ || !f_.getFullBelowCache()->touchIfExists(childHash.asUInt256()))
+        // The cache key carries the child's position beside its hash, so a hit answers for this
+        // child at this position and no other. The depth test runs first, so a child at kLeafDepth
+        // is judged by the arm below rather than answered from the cache: only a leaf sits there.
+        else if (
+            !backed_ || isLeafDepth(nodeID.getDepth() + 1) ||
+            !f_.getFullBelowCache()->touchIfExists(childHash.asUInt256(), nodeID, branch))
         {
             bool pending = false;
             auto d = descendAsync(
@@ -238,7 +240,28 @@ SHAMap::gmnProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
                 if (--mn.max <= 0)
                     return;
             }
-            else if (d->isInner() && !safeDowncast<SHAMapInnerNode*>(d)->isFullBelow(mn.generation))
+            // Only a leaf has a position of its own to judge, so the type is tested first. The
+            // depth test bounds the branch this arm names, and the arm below bounds the descent.
+            else if (
+                d->isLeaf() && !isLeafDepth(nodeID.getDepth()) && !belongsAt(nodeID, branch, *d))
+            {
+                // The judgment SHAMap::descend makes, for the descendAsync path, which hooks what
+                // it resolves before this runs. `fullBelow` is cleared as on the missing path.
+                fullBelow = false;
+                condemn(*d, childHash, nodeID.getChildNodeID(branch));
+                return;
+            }
+            else if (pastLeafDepth(nodeID.getDepth(), *d))
+            {
+                // A node resolved locally never passes through addKnownNode(), so the walk judges
+                // it here. The bound also covers nodeID, which the descent below writes.
+                condemn(*d, childHash, nodeID.getChildNodeID(branch));
+                return;
+            }
+            // The node's own full-below flag is not read here. The node object is shared by hash
+            // across maps and positions, so a flag another walk set says nothing about this
+            // position, and the position-keyed cache above has already missed for it.
+            else if (d->isInner())
             {
                 mn.stack.push(se);
 
@@ -257,10 +280,14 @@ SHAMap::gmnProcessNodes(MissingNodes& mn, MissingNodes::StackEntry& se)
 
     if (fullBelow)
     {  // No partial node encountered below this node
-        node->setFullBelowGen(mn.generation);
+        // The node flag is written for the root alone, since the root position is the only one a
+        // shared node object can vouch for; see SHAMapInnerNode::fullBelowGen_.
+        if (nodeID.isRoot())
+            node->setFullBelowGen(mn.generation);
         if (backed_)
         {
-            f_.getFullBelowCache()->insert(node->getHash().asUInt256());
+            // Keyed by position as well as hash, so the entry answers only for this position.
+            f_.getFullBelowCache()->insert(node->getHash().asUInt256(), nodeID);
         }
     }
 
@@ -291,6 +318,25 @@ SHAMap::gmnProcessDeferredReads(MissingNodes& mn)
         auto nodePtr = std::get<3>(deferredNode);
         auto const& nodeHash = parent->getChildHash(branch);
 
+        // A deferred entry carries the position the walk held when it posted the read, so a branch
+        // below it is named here only where the tree has room for one.
+        if (nodePtr && nodePtr->isLeaf() && !isLeafDepth(parentID.getDepth()) &&
+            !belongsAt(parentID, branch, *nodePtr))
+        {
+            // The judgment the synchronous paths make, for a node an async read resolved. Skips
+            // rather than returns, since every posted read must be drained while `mn` is alive.
+            condemn(*nodePtr, nodeHash, parentID.getChildNodeID(branch));
+            continue;
+        }
+
+        // The depth bound gmnProcessNodes carries beside its leaf test, for a node an async read
+        // resolved. Skips for the same reason as above.
+        if (nodePtr && pastLeafDepth(parentID.getDepth(), *nodePtr))
+        {
+            condemn(*nodePtr, nodeHash, parentID.getChildNodeID(branch));
+            continue;
+        }
+
         if (nodePtr)
         {  // Got the node
             nodePtr = parent->canonicalizeChild(branch, std::move(nodePtr));
@@ -301,6 +347,8 @@ SHAMap::gmnProcessDeferredReads(MissingNodes& mn)
         }
         else if ((mn.max > 0) && (mn.missingHashes.insert(nodeHash).second))
         {
+            // getChildNodeID() is safe here: gmnProcessNodes refuses to descend into an inner node
+            // at kLeafDepth, so a deferred parent sits at most one level above it.
             mn.missingNodes.emplace_back(parentID.getChildNodeID(branch), nodeHash.asUInt256());
             --mn.max;
         }
@@ -311,17 +359,20 @@ SHAMap::gmnProcessDeferredReads(MissingNodes& mn)
     mn.deferred = 0;
 }
 
-/**
- * Get a list of node IDs and hashes for nodes that are part of this SHAMap
- * but not available locally.  The filter can hold alternate sources of
- * nodes that are not permanently stored locally
- */
 std::vector<std::pair<SHAMapNodeID, UInt256>>
 SHAMap::getMissingNodes(int max, SHAMapSyncFilter const* filter)
 {
     XRPL_ASSERT(root_->getHash().isNonZero(), "xrpl::SHAMap::getMissingNodes : nonzero root hash");
     XRPL_ASSERT(max > 0, "xrpl::SHAMap::getMissingNodes : valid max input");
 
+    if (!isValid())
+    {
+        // The root node's own hash, since getHash() unshares the tree on a zero hash.
+        JLOG(journal_.warn()) << "getMissingNodes called on an invalid map, root hash "
+                              << root_->getHash() << " seq " << ledgerSeq();
+        return {};
+    }
+
     MissingNodes mn(
         max,
         filter,
@@ -354,6 +405,11 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter const* filter)
         {
             gmnProcessNodes(mn, pos);
 
+            // The walk just invalidated the map. The loop stops descending here but falls through
+            // to the drain below, since every posted read must be drained while `mn` is alive.
+            if (!isValid())
+                break;
+
             if (mn.max <= 0)
                 break;
 
@@ -383,6 +439,11 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter const* filter)
         if (mn.deferred != 0)
             gmnProcessDeferredReads(mn);
 
+        // Reads are drained, so the map can be abandoned. What was collected belongs to a tree
+        // that cannot exist.
+        if (!isValid())
+            return {};
+
         if (mn.max <= 0)
             return std::move(mn.missingNodes);
 
@@ -391,12 +452,10 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter const* filter)
 
             if (mn.stack.empty() && !mn.resumes.empty())
             {
-                // Recheck nodes we could not finish before
+                // Recheck nodes we could not finish before. A node's own full-below flag is not
+                // consulted: it is written for a root only, and a resumed node sits below one.
                 for (auto const& [innerNode, nodeId] : mn.resumes)
-                {
-                    if (!innerNode->isFullBelow(mn.generation))
-                        mn.stack.emplace(innerNode, nodeId, randInt(255), 0, true);
-                }
+                    mn.stack.emplace(innerNode, nodeId, randInt(255), 0, true);
 
                 mn.resumes.clear();
             }
@@ -416,6 +475,11 @@ SHAMap::getMissingNodes(int max, SHAMapSyncFilter const* filter)
 
     } while (node != nullptr);
 
+    // addKnownNode() on another thread can write the verdict after the loop's own test, so the map
+    // is judged once more before the result is returned.
+    if (!isValid())
+        return {};  // LCOV_EXCL_LINE: only that other thread reaches this, so no test does
+
     if (mn.missingNodes.empty())
         clearSynching();
 
@@ -527,11 +591,19 @@ SHAMap::addRootNode(
     XRPL_ASSERT(cowid_ >= 1, "xrpl::SHAMap::addRootNode : valid cowid");
     XRPL_ASSERT(rootNode, "xrpl::SHAMap::addRootNode : non-null root node");
 
-    // we already have a root_ node
+    // A map syncs against one hash and installs a root once, so a root already held is a duplicate
+    // only once it hashes to the hash asked for.
     if (root_->getHash().isNonZero())
     {
         JLOG(journal_.trace()) << "Got root node, already have one";
-        XRPL_ASSERT(root_->getHash() == hash, "xrpl::SHAMap::addRootNode : valid hash");
+
+        if (root_->getHash() != hash)
+        {
+            JLOG(journal_.warn()) << "Root node offered under hash " << hash
+                                  << ", but the map holds " << root_->getHash();
+            return SHAMapAddNode::invalid();
+        }
+
         return SHAMapAddNode::duplicate();
     }
 
@@ -555,7 +627,7 @@ SHAMap::addRootNode(
         Serializer s;
         root_->serializeWithPrefix(s);
         filter->gotNode(
-            false, root_->getHash(), ledgerSeq_, std::move(s.modData()), root_->getType());
+            false, root_->getHash(), ledgerSeq(), std::move(s.modData()), root_->getType());
     }
 
     return SHAMapAddNode::useful();
@@ -569,10 +641,6 @@ SHAMap::addKnownNode(
 {
     XRPL_ASSERT(!nodeID.isRoot(), "xrpl::SHAMap::addKnownNode : valid node");
     XRPL_ASSERT(treeNode, "xrpl::SHAMap::addKnownNode : non-null tree node");
-    XRPL_ASSERT_IF(
-        treeNode->isLeaf(),
-        nodeID.isPrefixOf(leafKey(*treeNode)),
-        "xrpl::SHAMap::addKnownNode : leaf position consistent with node ID");
 
     if (!isSynching())
     {
@@ -580,13 +648,13 @@ SHAMap::addKnownNode(
         return SHAMapAddNode::duplicate();
     }
 
-    auto const generation = f_.getFullBelowCache()->getGeneration();
     SHAMapNodeID currNodeID;
     auto currNode = root_.get();
 
-    while (currNode->isInner() &&
-           !safeDowncast<SHAMapInnerNode*>(currNode)->isFullBelow(generation) &&
-           (currNodeID.getDepth() < nodeID.getDepth()))
+    // The descent reads no node's own full-below flag: a node object is shared by hash across
+    // maps and positions, so that flag answers for the root position alone. The position-keyed
+    // cache lookup in the loop is the memo that answers for a position below it.
+    while (currNode->isInner() && (currNodeID.getDepth() < nodeID.getDepth()))
     {
         auto const branch = selectBranch(currNodeID, nodeID.getNodeID());
         auto inner = safeDowncast<SHAMapInnerNode*>(currNode);
@@ -598,7 +666,12 @@ SHAMap::addKnownNode(
         }
 
         auto childHash = inner->getChildHash(branch);
-        if (f_.getFullBelowCache()->touchIfExists(childHash.asUInt256()))
+
+        // The cache key carries the child's position beside its hash, so a hit answers for this
+        // child at this position and no other. The depth test runs first, so a node offered at
+        // kLeafDepth reaches the badDepth test below rather than being answered as a duplicate.
+        if (!isLeafDepth(currNodeID.getDepth() + 1) &&
+            f_.getFullBelowCache()->touchIfExists(childHash.asUInt256(), currNodeID, branch))
         {
             return SHAMapAddNode::duplicate();
         }
@@ -606,6 +679,14 @@ SHAMap::addKnownNode(
         auto prevNode = inner;
         std::tie(currNode, currNodeID) = descend(inner, currNodeID, branch, filter);
 
+        if (!isValid())
+        {
+            // descend condemned the map. `childHash` was read before that descent, so the
+            // comparison below would read a stale value.
+            JLOG(journal_.warn()) << "Node " << nodeID << " cannot be hooked into an invalid map";
+            return SHAMapAddNode::invalid();
+        }
+
         if (currNode != nullptr)
             continue;
 
@@ -616,25 +697,33 @@ SHAMap::addKnownNode(
             return SHAMapAddNode::invalid();
         }
 
-        // Inner nodes must be at a level strictly less than 64
-        // but leaf nodes (while notionally at level 64) can be
-        // at any depth up to and including 64:
-        if ((currNodeID.getDepth() > kLeafDepth) ||
-            (treeNode->isInner() && currNodeID.getDepth() == kLeafDepth))
+        // Every node from the root down hash-verified to get here, so the requested root hash
+        // itself commits to a shape no tree can have. The verdict belongs to that hash.
+        bool const badDepth = treeNode->isInner() && isLeafDepth(currNodeID.getDepth());
+        SOMETIMES(badDepth, "xrpl::SHAMap::addKnownNode : map is invalid");
+        if (badDepth)
         {
-            // Map is provably invalid
-            state_ = SHAMapState::Invalid;
-            return SHAMapAddNode::useful();
+            condemn(*treeNode, treeNode->getHash(), currNodeID);
+            return SHAMapAddNode::mapInvalidated();
         }
 
-        if (currNodeID != nodeID)
+        // The data hashes to the child at currNodeID but is labeled nodeID, so it is not the node
+        // asked for. Only the label is wrong, so the map stays sound.
+        bool const badPosition = (currNodeID != nodeID);
+        SOMETIMES(badPosition, "xrpl::SHAMap::addKnownNode : node ID does not match its position");
+        if (badPosition)
         {
-            // Either this node is broken or we didn't request it (yet)
-            JLOG(journal_.warn()) << "unable to hook node " << nodeID;
-            JLOG(journal_.info()) << " stuck at " << currNodeID;
-            JLOG(journal_.info()) << "got depth=" << nodeID.getDepth()
-                                  << ", walked to= " << currNodeID.getDepth();
-            return SHAMapAddNode::useful();
+            JLOG(journal_.warn()) << "Unable to hook node " << nodeID << ", stuck at "
+                                  << currNodeID;
+            return SHAMapAddNode::invalid();
+        }
+
+        // A leaf's own key names its position, and the hash test above ties this leaf to this
+        // parent, so the verdict belongs to the map. Below badPosition, since Invalid is terminal.
+        if (!belongsAt(nodeID, *treeNode))
+        {
+            condemn(*treeNode, treeNode->getHash(), nodeID);
+            return SHAMapAddNode::mapInvalidated();
         }
 
         if (backed_)
@@ -647,7 +736,7 @@ SHAMap::addKnownNode(
             Serializer s;
             treeNode->serializeWithPrefix(s);
             filter->gotNode(
-                false, childHash, ledgerSeq_, std::move(s.modData()), treeNode->getType());
+                false, childHash, ledgerSeq(), std::move(s.modData()), treeNode->getType());
         }
 
         return SHAMapAddNode::useful();
@@ -766,7 +855,7 @@ SHAMap::hasLeafNode(UInt256 const& tag, SHAMapHash const& targetNodeHash) const
         // Same kLeafDepth hazard as in visitDifferences above. That guard bounds the caller's own
         // traversal, not the map queried here, and the loop below descends from this map's root
         // independently, so this check is what keeps a malformed map from reaching getChildNodeID.
-        if (nodeID.getDepth() >= kLeafDepth)
+        if (isLeafDepth(nodeID.getDepth()))
         {
             // LCOV_EXCL_START
             UNREACHABLE("xrpl::SHAMap::hasLeafNode : inner node at leaf depth");
@@ -802,7 +891,7 @@ SHAMap::getProofPath(UInt256 const& key) const
         return {};
     }
 
-    if (auto const& node = stack.top().first; !node || node->isInner() ||
+    if (auto const& node = stack.top(); !node || node->isInner() ||
         intr_ptr::staticPointerCast<SHAMapLeafNode>(node)->peekItem()->key() != key)
     {
         JLOG(journal_.debug()) << "no path to " << key;
@@ -814,7 +903,7 @@ SHAMap::getProofPath(UInt256 const& key) const
     while (!stack.empty())
     {
         Serializer s;
-        stack.top().first->serializeForWire(s);
+        stack.top()->serializeForWire(s);
         path.emplace_back(std::move(s.modData()));
         stack.pop();
     }
@@ -849,8 +938,8 @@ SHAMap::verifyProofPath(UInt256 const& rootHash, UInt256 const& key, std::vector
                 // there. These nodes come off the wire, so a peer can still claim an inner one;
                 // reject it rather than passing this depth to selectBranch.
                 SOMETIMES(
-                    depth >= kLeafDepth, "xrpl::SHAMap::verifyProofPath : inner at leaf depth");
-                if (depth >= kLeafDepth)
+                    isLeafDepth(depth), "xrpl::SHAMap::verifyProofPath : inner at leaf depth");
+                if (isLeafDepth(depth))
                     return false;
 
                 auto nodeId = SHAMapNodeID::createID(depth, key);
@@ -859,10 +948,9 @@ SHAMap::verifyProofPath(UInt256 const& rootHash, UInt256 const& key, std::vector
             }
             else
             {
-                // The hash chain up to rootHash only proves this leaf sits where the path claims,
-                // not that it is the leaf for `key`: a peer could substitute any other leaf whose
-                // subtree hashes to the same value at every level above it. Checking the terminal
-                // leaf's own key is what ties the proof to `key` specifically.
+                // The hash chain up to rootHash proves this leaf sits where the path claims. Any
+                // leaf whose subtree hashes the same at every level above satisfies that chain, so
+                // the terminal leaf's own key is what ties the proof to `key`.
                 if (leafKey(*node) != key)
                     return false;
 

src/test/app/AcquireTestHelpers.h

@@ -0,0 +1,365 @@
+#pragma once
+
+#include <test/jtx/PeerStub.h>
+
+#include <xrpld/app/ledger/ConsensusTransSetSF.h>
+#include <xrpld/overlay/Peer.h>
+#include <xrpld/overlay/PeerSet.h>
+
+#include <xrpl/basics/SHAMapHash.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/resource/Charge.h>
+#include <xrpl/shamap/SHAMapAddNode.h>
+#include <xrpl/shamap/SHAMapNodeID.h>
+#include <xrpl/shamap/SHAMapTreeNode.h>
+
+#include <tests/libxrpl/shamap/DeepChain.h>
+
+#include <xrpl.pb.h>
+
+#include <atomic>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <set>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+namespace xrpl::test {
+
+// The chain builder needs only libxrpl, so it is shared with the gtest suites; see the header for
+// what keeps the protobuf reply builder below in this tree.
+using tests::DeepChain;
+
+// A smallest-possible leaf plus its 4-byte HashPrefix lands one byte short of the floor
+// ConsensusTransSetSF::gotNode() parses at, so a chain's leaf stays below the parse threshold.
+static_assert(
+    sizeof(std::uint32_t) + DeepChain::kLeafItemBytes < ConsensusTransSetSF::kMinTxNodeBytesToParse,
+    "a smallest-possible leaf must stay below the resubmission floor");
+
+/**
+ * A peer that records what it was charged. Every other method comes from
+ * PeerStub.
+ *
+ * One instance per packet keeps charges() unambiguous about which packet was
+ * charged what.
+ *
+ * charges_ is unguarded: charging happens on the packet path, so every charge
+ * lands on the thread that fed the packet in.
+ */
+class ChargeRecordingPeer : public PeerStub
+{
+public:
+    /**
+     * @param hasTxSet What hasTxSet() reports, which is how an acquisition
+     *        decides whether this peer is worth asking. Defaults to true, so a
+     *        peer handed straight to takeNodes() needs no argument.
+     */
+    explicit ChargeRecordingPeer(bool hasTxSet = true) : PeerStub(nextId()), hasTxSet_(hasTxSet)
+    {
+    }
+
+    void
+    charge(resource::Charge const& fee, std::string const&) override
+    {
+        charges_.push_back(fee);
+    }
+
+    /**
+     * @return Every fee this peer has been charged, in the order charged.
+     */
+    [[nodiscard]] std::vector<resource::Charge> const&
+    charges() const
+    {
+        return charges_;
+    }
+
+    // PeerStub returns false for both, and an acquisition asks the peers reporting true.
+
+    [[nodiscard]] bool
+    hasTxSet(UInt256 const&) const override
+    {
+        return hasTxSet_;
+    }
+
+    [[nodiscard]] bool
+    hasLedger(UInt256 const&, std::uint32_t) const override
+    {
+        return true;
+    }
+
+private:
+    /**
+     * The next id to hand out, distinct per instance because
+     * RequestCountingPeerSet dedups by tracked id and PeerStub's own default is
+     * zero for every instance.
+     *
+     * @return The id.
+     */
+    [[nodiscard]] static ID
+    nextId()
+    {
+        static std::atomic<ID> next{1};
+        return next++;
+    }
+
+    std::vector<resource::Charge> charges_;
+    bool hasTxSet_;
+};
+
+/**
+ * A peer set that counts the requests an acquisition makes through it. Offers
+ * peers to a hasItem/onPeerAdded callback pair, hard-filtered by hasItem (which
+ * only scores in the real peer set) and deduped by tracked id.
+ *
+ * A count is a call, not a delivery: a request naming no peer reaches nobody
+ * while this set tracks none. requests() and broadcasts() are counted apart.
+ *
+ * Every write here is guarded, since the retry timer drives addPeers() and
+ * sendRequest() from a job thread while the test reads the results.
+ */
+class RequestCountingPeerSet : public PeerSet
+{
+public:
+    /**
+     * @param candidates The peers addPeers() may offer, in the order they are
+     *        considered. Fixed at construction. Empty offers no one.
+     */
+    explicit RequestCountingPeerSet(std::vector<std::shared_ptr<Peer>> candidates = {})
+        : candidates_(std::move(candidates))
+    {
+    }
+
+    /**
+     * Offer the candidates to the caller, the way the real peer set offers the
+     * peers the overlay is tracking.
+     *
+     * @param limit The most peers to add, recorded for firstLimit().
+     * @param hasItem Hard-filters the candidates worth asking, where the real
+     *        peer set only scores with it.
+     * @param onPeerAdded Called for each selected candidate.
+     */
+    void
+    addPeers(
+        std::size_t limit,
+        std::function<bool(std::shared_ptr<Peer> const&)> hasItem,
+        std::function<void(std::shared_ptr<Peer> const&)> onPeerAdded) override
+    {
+        std::vector<std::shared_ptr<Peer>> selected;
+        {
+            std::scoped_lock const lock(mutex_);
+
+            if (!firstLimit_)
+                firstLimit_ = limit;
+
+            for (auto const& candidate : candidates_)
+            {
+                if (selected.size() >= limit)
+                    break;
+                // Dedup by tracked id, like the real peer set: onPeerAdded runs once per candidate.
+                if (hasItem(candidate) && addedPeers_.insert(candidate->id()).second)
+                    selected.push_back(candidate);
+            }
+        }
+
+        // Outside the lock: onPeerAdded() reenters this object through sendRequest().
+        for (auto const& peer : selected)
+            onPeerAdded(peer);
+    }
+
+    /**
+     * Record the request, and record it in the broadcast count as well when it
+     * names no peer.
+     *
+     * @param peer The peer to ask, or null to ask every tracked peer.
+     */
+    void
+    sendRequest(
+        ::google::protobuf::Message const&,
+        protocol::MessageType,
+        std::shared_ptr<Peer> const& peer) override
+    {
+        std::scoped_lock const lock(mutex_);
+        ++requests_;
+        if (!peer)
+            ++broadcasts_;
+    }
+
+    /**
+     * The ids of every peer addPeers() has selected, which is what an
+     * acquisition takes for the peers it is tracking.
+     *
+     * Unguarded: every caller of this and of addPeers() is an acquisition
+     * holding its own mtx_, so the ids stay fixed while a caller iterates. A
+     * test thread reads addedPeers() instead.
+     * InboundLedger::getPeerCount() reports zero for these, since it resolves
+     * ids through the overlay, while these peers live only in this harness.
+     *
+     * @return The ids.
+     */
+    [[nodiscard]] std::set<Peer::ID> const&
+    getPeerIds() const override
+    {
+        return addedPeers_;
+    }
+
+    /**
+     * @return How many requests have been sent through this peer set, counting
+     *         a broadcast as one.
+     */
+    [[nodiscard]] int
+    requests() const
+    {
+        std::scoped_lock const lock(mutex_);
+        return requests_;
+    }
+
+    /**
+     * How many of those requests carried no peer of their own.
+     *
+     * @return The count.
+     */
+    [[nodiscard]] int
+    broadcasts() const
+    {
+        std::scoped_lock const lock(mutex_);
+        return broadcasts_;
+    }
+
+    /**
+     * The limit the first addPeers() call asked for, which is init()'s, since
+     * onTimer() keeps calling addPeers(1) for as long as an acquisition runs.
+     *
+     * @return The limit, or nullopt if addPeers() has not been called.
+     */
+    [[nodiscard]] std::optional<std::size_t>
+    firstLimit() const
+    {
+        std::scoped_lock const lock(mutex_);
+        return firstLimit_;
+    }
+
+    /**
+     * A set, because onTimer() keeps re-offering the same candidates.
+     *
+     * @return The ids of every peer addPeers() has selected so far.
+     */
+    [[nodiscard]] std::set<Peer::ID>
+    addedPeers() const
+    {
+        std::scoped_lock const lock(mutex_);
+        return addedPeers_;
+    }
+
+private:
+    std::vector<std::shared_ptr<Peer>> const candidates_;
+
+    mutable std::mutex mutex_;
+    int requests_{0};
+    int broadcasts_{0};
+    std::optional<std::size_t> firstLimit_;
+    std::set<Peer::ID> addedPeers_;
+};
+
+/**
+ * The given nodes of a chain as a TMLedgerData, so a test can go through the
+ * real dispatch. Not in DeepChain, since the protobuf types are xrpld and that
+ * header is shared with the libxrpl-only gtest binary.
+ *
+ * @param chain The chain the nodes came from, which names the reply by default.
+ * @param data The nodes to include, each with its claimed position.
+ * @param type The reply type, which selects which map the receiver applies it
+ *        to.
+ * @param ledgerHash The hash the reply claims to be about, defaulting to the
+ *        chain root for a TX set. A ledger acquisition wants its header hash
+ *        here, since the chain root is only that ledger's account hash.
+ * @param ledgerSeq The sequence to name in the reply.
+ * @return The reply packet.
+ */
+[[nodiscard]] inline std::shared_ptr<protocol::TMLedgerData>
+packetFor(
+    DeepChain const& chain,
+    std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> const& data,
+    protocol::TMLedgerInfoType type = protocol::liTS_CANDIDATE,
+    std::optional<UInt256> const& ledgerHash = std::nullopt,
+    std::uint32_t ledgerSeq = 0)
+{
+    auto packet = std::make_shared<protocol::TMLedgerData>();
+    auto const hash = ledgerHash.value_or(chain.rootHash.asUInt256());
+    packet->set_ledgerhash(hash.data(), UInt256::size());
+    packet->set_ledgerseq(ledgerSeq);
+    packet->set_type(type);
+
+    for (auto const& [nodeID, node] : data)
+    {
+        Serializer s;
+        node->serializeForWire(s);
+
+        auto* const ledgerNode = packet->add_nodes();
+        ledgerNode->set_nodedata(s.peekData().data(), s.peekData().size());
+
+        // A leaf carries its own key, so the receiver rebuilds its position from that plus a
+        // depth. An inner node has no key and needs the full ID. The two fields are a oneof.
+        if (node->isLeaf())
+        {
+            ledgerNode->set_depth(nodeID.getDepth());
+        }
+        else
+        {
+            ledgerNode->set_id(nodeID.getRawString());
+        }
+    }
+
+    return packet;
+}
+
+/**
+ * Poll until the condition holds, or give up. An acquisition's own timer and
+ * the jobs it hands finished work to both run on other threads.
+ *
+ * @param condition What to wait for.
+ * @param deadline The longest to wait.
+ * @return Whether the condition held before the deadline.
+ */
+[[nodiscard]] inline bool
+waitFor(
+    std::function<bool()> const& condition,
+    std::chrono::steady_clock::duration deadline = std::chrono::seconds{10})
+{
+    auto const giveUp = std::chrono::steady_clock::now() + deadline;
+    while (std::chrono::steady_clock::now() < giveUp)
+    {
+        if (condition())
+            return true;
+        std::this_thread::sleep_for(std::chrono::milliseconds{10});
+    }
+    return condition();
+}
+
+/**
+ * Whether a batch verdict carries exactly the given counts.
+ *
+ * The counts, since get() is a log format. It is pinned once, in the
+ * SHAMapAddNode tests, and is what to pass BEAST_EXPECTS() as the reason a
+ * check here failed.
+ *
+ * @param san The verdict to check.
+ * @param good How many nodes the batch should have hooked in.
+ * @param bad How many it should have rejected.
+ * @param duplicate How many it should have already held.
+ * @return Whether the verdict matches.
+ */
+[[nodiscard]] inline bool
+tallyIs(SHAMapAddNode const& san, int good, int bad, int duplicate)
+{
+    return san.getGood() == good && san.getBad() == bad && san.getDuplicate() == duplicate;
+}
+
+}  // namespace xrpl::test

src/test/app/LedgerHistory_test.cpp

@@ -11,6 +11,7 @@
 
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/basics/chrono.h>
+#include <xrpl/basics/contract.h>
 #include <xrpl/beast/insight/NullCollector.h>
 #include <xrpl/beast/unit_test/suite.h>
 #include <xrpl/ledger/ApplyView.h>
@@ -22,6 +23,7 @@
 
 #include <cassert>
 #include <memory>
+#include <stdexcept>
 #include <vector>
 
 namespace xrpl::test {
@@ -70,11 +72,15 @@ public:
         }
         res->unshare();
 
-        // Accept ledger
-        res->setAccepted(
-            res->header().closeTime,
-            res->header().closeTimeResolution,
-            true /* close time correct*/);
+        // Accept the ledger. Thrown rather than asserted because a refusal leaves res unusable,
+        // and this helper is static, so BEAST_EXPECT is out of reach.
+        if (!res->setAccepted(
+                res->header().closeTime,
+                res->header().closeTimeResolution,
+                true /* close time correct*/))
+        {
+            Throw<std::runtime_error>("makeLedger: ledger could not be accepted");
+        }
         lh.insert(res, false);
         return res;
     }

src/test/app/RCLValidations_test.cpp

@@ -100,7 +100,7 @@ class RCLValidations_test : public beast::unit_test::Suite
             BEAST_EXPECT(next->read(keylet::feeSettings()));
             if (forceHash)
             {
-                next->setImmutable();
+                BEAST_EXPECT(next->setImmutable());
                 forceHash = false;
             }
 

src/test/rpc/BookChanges_test.cpp

@@ -13,6 +13,7 @@
 
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/basics/chrono.h>
+#include <xrpl/basics/contract.h>
 #include <xrpl/beast/hash/uhash.h>
 #include <xrpl/beast/unit_test/suite.h>
 #include <xrpl/json/json_value.h>
@@ -190,7 +191,7 @@ public:
             metadata->add(*metaSerializer);
 
             ledger->rawTxInsert(UInt256{1}, txSerializer, metaSerializer);
-            ledger->setImmutable();
+            BEAST_EXPECT(ledger->setImmutable());
             ledger->setValidated();
 
             try
@@ -260,7 +261,12 @@ public:
             ledger->rawTxInsert(UInt256{seq}, txSerializer, metaSerializer);
         }
 
-        ledger->setImmutable();
+        // Thrown rather than asserted because a refusal leaves the ledger unusable, and this
+        // helper is static, so BEAST_EXPECT is out of reach.
+        if (!ledger->setImmutable())
+        {
+            Throw<std::runtime_error>("bookChangesFor: ledger could not be made immutable");
+        }
         ledger->setValidated();
 
         return xrpl::rpc::computeBookChanges(std::static_pointer_cast<Ledger const>(ledger));

src/tests/libxrpl/helpers/TestFamily.h

@@ -1,123 +0,0 @@
-#pragma once
-
-#include <xrpl/basics/ByteUtilities.h>
-#include <xrpl/basics/base_uint.h>
-#include <xrpl/basics/chrono.h>
-#include <xrpl/basics/contract.h>
-#include <xrpl/beast/utility/Journal.h>
-#include <xrpl/config/BasicConfig.h>
-#include <xrpl/config/Constants.h>
-#include <xrpl/nodestore/Database.h>
-#include <xrpl/nodestore/DummyScheduler.h>
-#include <xrpl/nodestore/Manager.h>
-#include <xrpl/shamap/Family.h>
-#include <xrpl/shamap/FullBelowCache.h>
-#include <xrpl/shamap/TreeNodeCache.h>
-
-#include <cstdint>
-#include <memory>
-#include <stdexcept>
-
-namespace xrpl::test {
-
-/**
- * Test implementation of Family for unit tests.
- *
- * Uses an in-memory NodeStore database and simple caches.
- * The missingNode methods throw since tests shouldn't encounter missing nodes.
- */
-class TestFamily : public Family
-{
-private:
-    std::unique_ptr<node_store::Database> db_;
-    TestStopwatch clock_;
-    std::shared_ptr<FullBelowCache> fbCache_;
-    std::shared_ptr<TreeNodeCache> tnCache_;
-    node_store::DummyScheduler scheduler_;
-    beast::Journal j_;
-
-public:
-    explicit TestFamily(beast::Journal j)
-        : fbCache_(std::make_shared<FullBelowCache>("TestFamily full below cache", clock_, j))
-        , tnCache_(
-              std::make_shared<TreeNodeCache>(
-                  "TestFamily tree node cache",
-                  65536,
-                  std::chrono::minutes{1},
-                  clock_,
-                  j))
-        , j_(j)
-    {
-        Section config;
-        config.set(Keys::kType, "memory");
-        config.set(Keys::kPath, "TestFamily");
-        db_ = node_store::Manager::instance().makeDatabase(megabytes(4), scheduler_, 1, config, j);
-    }
-
-    node_store::Database&
-    db() override
-    {
-        return *db_;
-    }
-
-    [[nodiscard]] node_store::Database const&
-    db() const override
-    {
-        return *db_;
-    }
-
-    beast::Journal const&
-    journal() override
-    {
-        return j_;
-    }
-
-    std::shared_ptr<FullBelowCache>
-    getFullBelowCache() override
-    {
-        return fbCache_;
-    }
-
-    std::shared_ptr<TreeNodeCache>
-    getTreeNodeCache() override
-    {
-        return tnCache_;
-    }
-
-    void
-    sweep() override
-    {
-        fbCache_->sweep();
-        tnCache_->sweep();
-    }
-
-    void
-    missingNodeAcquireBySeq(std::uint32_t refNum, UInt256 const& nodeHash) override
-    {
-        Throw<std::runtime_error>("TestFamily: missing node (by seq)");
-    }
-
-    void
-    missingNodeAcquireByHash(UInt256 const& refHash, std::uint32_t refNum) override
-    {
-        Throw<std::runtime_error>("TestFamily: missing node (by hash)");
-    }
-
-    void
-    reset() override
-    {
-        (*fbCache_).reset();
-        (*tnCache_).reset();
-    }
-
-    /**
-     * Access the test clock for time manipulation in tests.
-     */
-    TestStopwatch&
-    clock()
-    {
-        return clock_;
-    }
-};
-
-}  // namespace xrpl::test

src/tests/libxrpl/helpers/TestServiceRegistry.h

@@ -12,8 +12,8 @@
 
 #include <boost/asio/io_context.hpp>
 
-#include <helpers/TestFamily.h>
 #include <helpers/TestSink.h>
+#include <shamap/common.h>
 
 #include <cstdint>
 #include <memory>
@@ -73,7 +73,7 @@ class TestServiceRegistry : public ServiceRegistry
 {
     TestLogs logs_{beast::Severity::Warning};
     boost::asio::io_context ioContext_;
-    TestFamily family_{logs_.journal("TestFamily")};
+    tests::TestNodeFamily family_{logs_.journal("TestNodeFamily")};
     LoadFeeTrack feeTrack_{logs_.journal("LoadFeeTrack")};
     TestNetworkIDService networkIDService_;
     HashRouter hashRouter_{HashRouter::Setup{}, stopwatch()};

src/tests/libxrpl/helpers/TxTest.cpp

@@ -207,7 +207,10 @@ TxTest::close()
         accum.apply(*newLedger);
     }
 
-    newLedger->setAccepted(ledgerCloseTime, newLedger->header().closeTimeResolution, true);
+    if (!newLedger->setAccepted(ledgerCloseTime, newLedger->header().closeTimeResolution, true))
+    {
+        Throw<std::runtime_error>("TxTest::close: ledger has an invalid map");
+    }
 
     closedLedger_ = newLedger;
 

src/tests/libxrpl/shamap/DeepChain.h

@@ -0,0 +1,320 @@
+#pragma once
+
+#include <xrpl/basics/Blob.h>
+#include <xrpl/basics/SHAMapHash.h>
+#include <xrpl/basics/Slice.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/contract.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/shamap/SHAMap.h>
+#include <xrpl/shamap/SHAMapAddNode.h>
+#include <xrpl/shamap/SHAMapLeafNode.h>
+#include <xrpl/shamap/SHAMapNodeID.h>
+#include <xrpl/shamap/SHAMapTreeNode.h>
+
+#include <cstddef>
+#include <optional>
+#include <stdexcept>
+#include <utility>
+#include <vector>
+
+namespace xrpl::tests {
+
+/**
+ * A chain of inner nodes in wire form, from the root down to one deepest node,
+ * each with one real child and, under withDecoys(), an unresolvable second one.
+ *
+ * Built bottom-up, so every node hashes correctly and the root hash commits to
+ * the whole shape. Each node sits on the branch pathKey selects at its depth.
+ *
+ * Two shapes: the constructors run inner nodes all the way to
+ * SHAMap::kLeafDepth, a depth only a leaf may occupy, and toLeaf() ends at a
+ * real transaction leaf.
+ *
+ * Depends only on libxrpl, so either test tree can include it. The xrpld
+ * counterpart is test::packetFor() in src/test/app/AcquireTestHelpers.h.
+ */
+struct DeepChain
+{
+    // nodes[d] is the deserialized node for depth d.
+    std::vector<SHAMapTreeNodePtr> nodes;
+    SHAMapHash rootHash;
+
+    // The key whose path through the tree this chain spells out. Zero for a chain
+    // built without a leaf, which therefore sits on branch 0 at every depth.
+    UInt256 pathKey;
+
+    // The depth of the deepest node, the last one nodesBelowRoot() hands out.
+    unsigned int deepestDepth{SHAMap::kLeafDepth};
+
+    /**
+     * The payload size of the leaf toLeaf() builds, which is the smallest a
+     * SHAMap item may be.
+     */
+    static constexpr std::size_t kLeafItemBytes = kMinShaMapItemBytes;
+
+    /**
+     * A chain of inner nodes reaching SHAMap::kLeafDepth, a depth only a leaf
+     * may occupy.
+     *
+     * @param seed Varies the chain's nodes. Two chains built from one seed hold
+     *        the same nodes, which caches and fetch packs key by hash.
+     */
+    explicit DeepChain(unsigned int seed = 1) : DeepChain(std::nullopt, seed, Decoy::No)
+    {
+    }
+
+    /**
+     * The same chain, with an unresolvable second child at every level, so a
+     * backed map's descendAsync() posts a real asynchronous read per level.
+     *
+     * Offered only for this shape: the decoy sits on branch 1, which is free
+     * only while pathKey is zero.
+     *
+     * @param seed Varies the whole chain. See the constructor.
+     * @return The chain.
+     */
+    [[nodiscard]] static DeepChain
+    withDecoys(unsigned int seed = 1)
+    {
+        return DeepChain{std::nullopt, seed, Decoy::Yes};
+    }
+
+    /**
+     * A chain ending in a real transaction leaf, which completes an
+     * acquisition.
+     *
+     * @param depth Where the leaf sits, at most SHAMap::kLeafDepth. Zero puts
+     *        the leaf at the root. A deeper value throws std::logic_error, since
+     *        no leaf can sit there.
+     * @param seed Varies the leaf's contents, and so the whole chain. See the
+     *        constructor.
+     * @return The chain.
+     */
+    [[nodiscard]] static DeepChain
+    toLeaf(unsigned int depth, unsigned int seed = 1)
+    {
+        if (depth > SHAMap::kLeafDepth)
+            Throw<std::logic_error>("DeepChain: leaf depth past SHAMap::kLeafDepth");
+        return DeepChain{std::optional{depth}, seed, Decoy::No};
+    }
+
+    /**
+     * The node the chain holds at the given depth, root first.
+     *
+     * @param depth The depth of the node to return, at most deepestDepth.
+     * @return The node.
+     */
+    [[nodiscard]] SHAMapTreeNodePtr
+    nodeAt(unsigned int depth) const
+    {
+        return nodes[depth];
+    }
+
+    /**
+     * Where the node at the given depth claims to belong, which is on the
+     * path to pathKey.
+     *
+     * @param depth The depth of the node to locate.
+     * @return The node's claimed position.
+     */
+    [[nodiscard]] SHAMapNodeID
+    idAt(unsigned int depth) const
+    {
+        return SHAMapNodeID::createID(depth, pathKey);
+    }
+
+    /**
+     * The same node in the prefixed form used for storage and fetch packs,
+     * which is what hashes to the node's own hash.
+     *
+     * @param depth The depth of the node to serialize.
+     * @return The node's prefixed serialized form.
+     */
+    [[nodiscard]] Blob
+    prefixedNodeAt(unsigned int depth) const
+    {
+        Serializer s;
+        nodeAt(depth)->serializeWithPrefix(s);
+        return s.modData();
+    }
+
+    /**
+     * Every node below the root, down to and including the deepest one.
+     *
+     * @param firstDepth The shallowest node to include, so a caller can feed
+     *        the chain in more than one batch.
+     * @return The nodes, each with its claimed position.
+     */
+    [[nodiscard]] std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>>
+    nodesBelowRoot(unsigned int firstDepth = 1) const
+    {
+        std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data;
+        for (auto depth = firstDepth; depth <= deepestDepth; ++depth)
+            data.emplace_back(idAt(depth), nodeAt(depth));
+        return data;
+    }
+
+    /**
+     * Every node in the chain, root first.
+     *
+     * @return The nodes, each with its claimed position.
+     */
+    [[nodiscard]] std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>>
+    allNodes() const
+    {
+        return nodesBelowRoot(0);
+    }
+
+    /**
+     * Fill a synching map, stopping one level short of the deepest node so the
+     * caller offers that one itself.
+     *
+     * Returns rather than asserts, since two test frameworks share this header.
+     *
+     * @param map The map to fill.
+     * @return Whether the root and every node above the deepest one was
+     *         accepted.
+     */
+    [[nodiscard]] bool
+    fill(SHAMap& map) const
+    {
+        if (!map.addRootNode(rootHash, nodeAt(0), nullptr).isGood())
+            return false;
+
+        for (auto depth = 1u; depth < deepestDepth; ++depth)
+        {
+            if (!map.addKnownNode(idAt(depth), nodeAt(depth), nullptr).isUseful())
+                return false;
+        }
+
+        return true;
+    }
+
+    /**
+     * Offer the deepest node, which for a fabricated chain is the inner node at
+     * SHAMap::kLeafDepth, a depth only a leaf may occupy.
+     *
+     * @param map The map to offer the node to, filled by fill() first.
+     * @return The verdict addKnownNode() reached.
+     */
+    [[nodiscard]] SHAMapAddNode
+    addOffendingNode(SHAMap& map) const
+    {
+        return map.addKnownNode(idAt(deepestDepth), nodeAt(deepestDepth), nullptr);
+    }
+
+private:
+    // Whether each level carries a second child that stays unresolvable.
+    enum class Decoy { No, Yes };
+
+    /**
+     * Build any of the shapes.
+     *
+     * @param leafDepth Where a real transaction leaf sits, or nullopt to run
+     *        inner nodes all the way to SHAMap::kLeafDepth instead.
+     * @param seed Varies the chain's contents. See the public entry points.
+     * @param decoy Whether every level carries an unresolvable second child.
+     */
+    DeepChain(std::optional<unsigned int> leafDepth, unsigned int seed, Decoy decoy)
+        : nodes(leafDepth.value_or(SHAMap::kLeafDepth) + 1)
+    {
+        if (!leafDepth)
+        {
+            // With no leaf depth given, the deepest inner node points at a child that stays
+            // unresolvable.
+            buildInnersDownTo(SHAMap::kLeafDepth, SHAMapHash{UInt256{seed}}, decoy);
+            return;
+        }
+
+        // Exactly kLeafItemBytes of payload, the smallest a leaf item may be. Checked rather than
+        // assumed, since a caller relates that constant to a threshold of its own.
+        Serializer payload;
+        payload.add32(seed);
+        payload.add32(0);
+        payload.add32(0);
+        if (payload.size() != kLeafItemBytes)
+            Throw<std::logic_error>("DeepChain: unexpected leaf payload size");
+
+        Serializer wire;
+        wire.addRaw(payload.peekData());
+        wire.add8(kWireTypeTransaction);
+
+        auto const leaf = SHAMapTreeNode::makeFromWire(makeSlice(wire.peekData()));
+
+        // A transaction leaf's key is the hash of its own contents, so its position follows
+        // this key's nibbles.
+        pathKey = leafKey(*leaf);
+        deepestDepth = *leafDepth;
+        nodes[*leafDepth] = leaf;
+
+        if (*leafDepth == 0)
+        {
+            // The leaf is the root, so the build stops here.
+            rootHash = leaf->getHash();
+            return;
+        }
+
+        buildInnersDownTo(*leafDepth - 1, leaf->getHash(), decoy);
+    }
+
+    /**
+     * Fill in inner nodes from the root down to the given depth and record the
+     * root hash. Each carries one real child and, under Decoy::Yes, an
+     * unresolvable second one.
+     *
+     * Bottom-up, since each node's hash covers the child hash below it.
+     *
+     * @param deepest The depth of the deepest inner node to build. May be
+     *        SHAMap::kLeafDepth.
+     * @param childHash What that deepest inner node points at.
+     * @param decoy Whether to add an unresolvable second child at every level.
+     */
+    void
+    buildInnersDownTo(unsigned int deepest, SHAMapHash childHash, Decoy decoy)
+    {
+        // The decrement is in the body rather than the condition, so the counter never steps
+        // below zero. UndefinedBehaviorSanitizer reports that wraparound, and the project halts
+        // on its first report.
+        for (auto depth = deepest + 1; depth > 0;)
+        {
+            --depth;
+
+            // A key has 64 nibbles, so SHAMap::kLeafDepth is one past the last nibble
+            // selectBranch() reads from a 32-byte key. Such a chain has a zero pathKey, so branch
+            // 0 is the position it claims.
+            auto const branch = depth == SHAMap::kLeafDepth ? 0u : selectBranch(depth, pathKey);
+
+            Serializer s;
+            s.addBitString(childHash.asUInt256());
+            s.add8(static_cast<unsigned char>(branch));
+
+            if (decoy == Decoy::Yes)
+            {
+                // The decoy sits at branch 1, which is free while pathKey is zero, so the
+                // compressed-inner-node parser sees two distinct branches.
+                if (branch == 1)
+                    Throw<std::logic_error>("DeepChain: decoy branch collides with real child");
+
+                // Derived from the depth, so each level posts its own read. No node stands behind
+                // this hash, so a read for it stays outstanding.
+                UInt256 decoyHash;
+                decoyHash.begin()[0] = 0xDE;
+                decoyHash.begin()[1] = 0xC0;
+                decoyHash.begin()[2] = static_cast<unsigned char>(depth);
+                s.addBitString(decoyHash);
+                s.add8(1);
+            }
+
+            s.add8(kWireTypeCompressedInner);
+
+            auto node = SHAMapTreeNode::makeFromWire(makeSlice(s.peekData()));
+            childHash = node->getHash();
+            nodes[depth] = std::move(node);
+        }
+
+        rootHash = childHash;
+    }
+};
+
+}  // namespace xrpl::tests

src/tests/libxrpl/shamap/InnerNode.h

@@ -0,0 +1,100 @@
+#pragma once
+
+#include <xrpl/basics/SHAMapHash.h>
+#include <xrpl/basics/Slice.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/contract.h>
+#include <xrpl/protocol/Serializer.h>
+#include <xrpl/shamap/SHAMapInnerNode.h>
+#include <xrpl/shamap/SHAMapTreeNode.h>
+
+#include <array>
+#include <cstdint>
+#include <stdexcept>
+#include <vector>
+
+namespace xrpl::tests {
+
+/**
+ * One child of an inner node a test assembles by hand.
+ */
+struct InnerChild
+{
+    unsigned int branch{};
+    SHAMapHash hash;
+};
+
+/**
+ * Assemble an inner node in the wire format's full form.
+ *
+ * A full inner node is all 16 branch hashes back to back in branch order,
+ * followed by the wire type byte. A branch absent from `children`, or given a
+ * zero hash, yields an empty branch, since the parser derives which branches
+ * exist from which hashes are non-zero. The node's hash is already correct,
+ * since makeFromWire() computes it.
+ *
+ * @param children The branch and hash of each child to record. Throws if a
+ *        branch is at or above SHAMapInnerNode::kBranchFactor, or if one
+ *        branch is named twice.
+ * @return The node, or nullptr if the bytes do not parse.
+ */
+[[nodiscard]] inline SHAMapTreeNodePtr
+makeFullInnerNode(std::vector<InnerChild> const& children)
+{
+    // The full form carries a slot for every branch, so a child's position comes from the slot it
+    // is written to.
+    std::array<UInt256, SHAMapInnerNode::kBranchFactor> hashes{};
+
+    // A duplicate branch is refused here, and one past the last to match the compressed parser.
+    std::uint32_t seen = 0;
+    for (auto const& child : children)
+    {
+        if (child.branch >= SHAMapInnerNode::kBranchFactor)
+            Throw<std::logic_error>("makeFullInnerNode: branch is past the last one");
+
+        auto const bit = 1u << child.branch;
+        if ((seen & bit) != 0)
+            Throw<std::logic_error>("makeFullInnerNode: branch named twice");
+        seen |= bit;
+
+        hashes.at(child.branch) = child.hash.asUInt256();
+    }
+
+    Serializer s;
+    for (auto const& hash : hashes)
+        s.addBitString(hash);
+    s.add8(kWireTypeInner);
+
+    return SHAMapTreeNode::makeFromWire(makeSlice(s.peekData()));
+}
+
+/**
+ * Assemble an inner node in the wire format's compressed form.
+ *
+ * A compressed inner node is one 33-byte chunk per child, each a hash followed
+ * by the branch it sits on, and then the wire type byte. Its hash is already
+ * correct, for the same reason makeFullInnerNode()'s is.
+ *
+ * Nothing is checked here, unlike in makeFullInnerNode(). The branch travels as
+ * one byte, so any value up to 255 reaches the parser, which refuses a branch
+ * at or above SHAMapInnerNode::kBranchFactor and lets a repeated branch
+ * overwrite the hash recorded for it.
+ *
+ * @param children The branch and hash of each child to record.
+ * @return The node, or nullptr if the bytes do not parse.
+ */
+[[nodiscard]] inline SHAMapTreeNodePtr
+makeCompressedInnerNode(std::vector<InnerChild> const& children)
+{
+    Serializer s;
+    for (auto const& child : children)
+    {
+        s.addBitString(child.hash.asUInt256());
+        s.add8(static_cast<unsigned char>(child.branch));
+    }
+    s.add8(kWireTypeCompressedInner);
+
+    return SHAMapTreeNode::makeFromWire(makeSlice(s.peekData()));
+}
+
+}  // namespace xrpl::tests

src/tests/libxrpl/shamap/SHAMap.cpp

@@ -7,23 +7,30 @@
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/Zero.h>
+#include <xrpl/nodestore/Database.h>
+#include <xrpl/nodestore/NodeObject.h>
 #include <xrpl/protocol/Serializer.h>
+#include <xrpl/shamap/Family.h>
 #include <xrpl/shamap/SHAMapInnerNode.h>
 #include <xrpl/shamap/SHAMapItem.h>
 #include <xrpl/shamap/SHAMapLeafNode.h>
 #include <xrpl/shamap/SHAMapMissingNode.h>
 #include <xrpl/shamap/SHAMapNodeID.h>
+#include <xrpl/shamap/SHAMapSyncFilter.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
 #include <gtest/gtest.h>
 #include <helpers/TestSink.h>
+#include <shamap/InnerNode.h>
 #include <shamap/common.h>
 
 #include <algorithm>
 #include <array>
 #include <cstddef>
 #include <cstdint>
+#include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <type_traits>
@@ -229,8 +236,11 @@ TEST_P(SHAMapTest, add_traverse_snapshot_build_tear_and_iterate)
             EXPECT_EQ(map.getHash().asUInt256(), kHashes[k]);
             map.invariants();
         }
-        for (std::size_t k = kKeys.size(); k-- > 0;)
+        // The decrement is in the body, so the counter never steps below zero.
+        // UndefinedBehaviorSanitizer reports that wraparound.
+        for (std::size_t k = kKeys.size(); k > 0;)
         {
+            --k;
             EXPECT_EQ(map.getHash().asUInt256(), kHashes[k]);
             EXPECT_TRUE(map.delItem(kKeys[k]));
             map.invariants();
@@ -272,9 +282,8 @@ INSTANTIATE_TEST_SUITE_P(
     ::testing::Values(kBackedMode, kUnbackedMode),
     shamapBackingModeName);
 
-// Exercises the traversal stacks built by belowHelper. Each stack entry pairs a node with the ID
-// naming its position, and SHAMap asserts that pairing on every push, so these traversals fail
-// loudly in a Debug build if a node ID is ever derived from the wrong branch.
+// Exercises the traversal paths built by belowHelper. A path names each node's position by its own
+// length, and every push requires a leaf's key to lie under the branch it was reached through.
 class SHAMapTraversal : public ::testing::Test
 {
 protected:
@@ -455,15 +464,64 @@ TEST_F(SHAMapTraversal, bounds_agree_with_iteration_for_absent_keys)
     }
 }
 
+TEST_F(SHAMapTraversal, update_give_item_on_absent_key_returns_false)
+{
+    tests::TestNodeFamily f{j_};
+    auto keys = deepFanOutKeys();
+    SHAMap map{SHAMapType::FREE, f};
+    fillMap(map, keys);
+
+    // A key absent from a fanned-out map, so the walk ends on an inner node with an empty branch.
+    auto const absentKey = UInt256{std::string_view{std::string(64, '0')}};
+    Buffer vuc{32};
+    std::fill_n(vuc.data(), vuc.size(), std::uint8_t{2});
+    EXPECT_FALSE(map.updateGiveItem(
+        SHAMapNodeType::TnAccountState, makeShamapitem(absentKey, std::move(vuc))));
+
+    // A single-item map probed with a key that selects the same root branch, so the walk ends on
+    // the leaf it reached, whose key is not the one asked for.
+    SHAMap single{SHAMapType::FREE, f};
+    fillMap(single, {keys.front()});
+
+    auto probe = keys.front();
+    std::fill_n(probe.begin() + 1, probe.size() - 1, std::uint8_t{0});
+    ASSERT_NE(probe, keys.front());
+
+    Buffer other{32};
+    std::fill_n(other.data(), other.size(), std::uint8_t{3});
+    EXPECT_FALSE(single.updateGiveItem(
+        SHAMapNodeType::TnAccountState, makeShamapitem(probe, std::move(other))));
+}
+
+TEST_F(SHAMapTraversal, del_item_on_absent_key_returns_false)
+{
+    tests::TestNodeFamily f{j_};
+    auto keys = deepFanOutKeys();
+    SHAMap map{SHAMapType::FREE, f};
+    fillMap(map, keys);
+
+    auto const before = map.getHash();
+
+    // Every key of this map shares its first nibble, so the root leaves the branch this key selects
+    // empty and the walk ends with an inner node on top of the path rather than a leaf.
+    auto const absentKey = UInt256{std::string_view{std::string(64, '0')}};
+    EXPECT_FALSE(map.delItem(absentKey));
+
+    // The refusal leaves the map as it was, which an answer of false alone does not say.
+    EXPECT_EQ(map.getHash(), before);
+    map.invariants();
+    for (auto const& key : keys)
+        EXPECT_TRUE(map.hasItem(key)) << to_string(key);
+}
+
 TEST_F(SHAMapTraversal, bounds_on_empty_map_return_end)
 {
     tests::TestNodeFamily f{j_};
     SHAMap map{SHAMapType::FREE, f};
     map.setUnbacked();
 
-    // The root is a childless inner node, so boundHelper's inner-node branch scans every branch on
-    // the requested side of the one id selects, finds them all empty, and falls through to end()
-    // rather than dereference a child.
+    // An empty map still leaves its root on the path, so boundHelper reads a non-empty path and
+    // answers end() from it.
     EXPECT_EQ(map.upperBound(UInt256{}), map.end());
     EXPECT_EQ(map.lowerBound(UInt256{}), map.end());
 
@@ -804,7 +862,7 @@ TEST_F(SHAMapPathProof, verify_proof_path)
 
 // A legitimate proof path for two keys sharing all 63 leading nibbles is 65 elements: inner nodes
 // at depths 0..63 plus the leaf at depth 64. This pins that the 65 bound is real, so the fix for
-// the forged-path case below must not simply tighten the length limit.
+// the built-path case below must not simply tighten the length limit.
 TEST_F(SHAMapPathProof, legitimate_deep_path_is_sixty_five_elements)
 {
     tests::TestNodeFamily f{j_};
@@ -837,7 +895,7 @@ TEST_F(SHAMapPathProof, legitimate_deep_path_is_sixty_five_elements)
     // NOLINTEND(bugprone-unchecked-optional-access)
 }
 
-// A forged path of 65 hash-chained inner nodes reaches depth kLeafDepth, where only the leaf
+// A built path of 65 hash-chained inner nodes reaches depth kLeafDepth, where only the leaf
 // terminating the path may sit. Such a path must be rejected.
 TEST_F(SHAMapPathProof, all_inner_path_at_leaf_depth_is_rejected)
 {
@@ -849,8 +907,11 @@ TEST_F(SHAMapPathProof, all_inner_path_at_leaf_depth_is_rejected)
     std::vector<Blob> path;
     SHAMapHash childHash{UInt256{1}};
 
-    for (auto depth = SHAMap::kLeafDepth + 1u; depth-- > 0;)
+    // The decrement is in the body, so the counter never steps below zero.
+    // UndefinedBehaviorSanitizer reports that wraparound.
+    for (auto depth = SHAMap::kLeafDepth + 1u; depth > 0;)
     {
+        --depth;
         auto const id = SHAMapNodeID::createID(std::min(depth, SHAMap::kLeafDepth - 1u), kTestKey);
         auto const branch = selectBranch(id, kTestKey);
 
@@ -871,25 +932,25 @@ TEST_F(SHAMapPathProof, all_inner_path_at_leaf_depth_is_rejected)
 }
 
 /**
- * Wrap a leaf blob in a forged root inner node whose branch for `key` carries that leaf's hash.
+ * Wrap a leaf blob in a built root inner node whose branch for `key` carries that leaf's hash.
  *
  * The resulting two-element path hash-chains for `key` no matter which leaf sits at the bottom,
  * which is exactly the substitution a peer could attempt.
  *
  * @param leafBlob the wire form of the leaf to place at the bottom of the path.
- * @param key the key the forged path claims to prove.
- * @return the path (deepest element first) and the forged root hash, or an empty path if the leaf
+ * @param key the key the built path claims to prove.
+ * @return the path (deepest element first) and the built root hash, or an empty path if the leaf
  *         blob does not parse.
  */
 static std::pair<std::vector<Blob>, UInt256>
-forgeRootOverLeaf(Blob const& leafBlob, UInt256 const& key)
+buildRootOverLeaf(Blob const& leafBlob, UInt256 const& key)
 {
     auto leaf = SHAMapTreeNode::makeFromWire(makeSlice(leafBlob));
     if (!leaf || !leaf->isLeaf())
         return {};
     leaf->updateHash();
 
-    auto const branch = selectBranch(SHAMapNodeID::createID(0, key), key);
+    auto const branch = selectBranch(0u, key);
     Serializer s;
     for (auto i = 0u; i < SHAMap::kBranchFactor; ++i)
         s.addBitString(i == branch ? leaf->getHash().asUInt256() : UInt256{});
@@ -904,7 +965,7 @@ forgeRootOverLeaf(Blob const& leafBlob, UInt256 const& key)
 }
 
 // The hash chain above a leaf proves nothing about which key that leaf holds, so a peer can graft a
-// genuine leaf from elsewhere in the map onto a path forged for another key. Comparing the terminal
+// genuine leaf from elsewhere in the map onto a path built for another key. Comparing the terminal
 // leaf's own key against the key being proved is what rejects it.
 TEST_F(SHAMapPathProof, substituted_leaf_for_other_key_is_rejected)
 {
@@ -934,17 +995,716 @@ TEST_F(SHAMapPathProof, substituted_leaf_for_other_key_is_rejected)
     auto const& otherLeaf = otherPath->front();
     // NOLINTEND(bugprone-unchecked-optional-access)
 
-    // Control: the forged root is accepted when the leaf below it really is kKey's leaf, so the
+    // Control: the built root is accepted when the leaf below it really is kKey's leaf, so the
     // rejection below can only come from the leaf key comparison.
-    auto const [goodPath, goodRoot] = forgeRootOverLeaf(ownLeaf, kKey);
+    auto const [goodPath, goodRoot] = buildRootOverLeaf(ownLeaf, kKey);
     ASSERT_EQ(goodPath.size(), 2u);
     EXPECT_TRUE(SHAMap::verifyProofPath(goodRoot, kKey, goodPath));
 
-    // Same forged root, but kOtherKey's leaf substituted at the bottom: the hash chain still
+    // Same built root, but kOtherKey's leaf substituted at the bottom: the hash chain still
     // validates, yet the path does not prove anything about kKey.
-    auto const [badPath, badRoot] = forgeRootOverLeaf(otherLeaf, kKey);
+    auto const [badPath, badRoot] = buildRootOverLeaf(otherLeaf, kKey);
     ASSERT_EQ(badPath.size(), 2u);
     EXPECT_FALSE(SHAMap::verifyProofPath(badRoot, kKey, badPath));
 }
 
+/**
+ * Run `call` and check it throws SHAMapMissingNode naming a key, not a hash.
+ *
+ * A walk that decides it cannot go on names the key it was searching for, while
+ * a failure to fetch names the hash. Both throw the same type, so the payload
+ * is what tells the two apart.
+ *
+ * @param call the call expected to throw.
+ */
+template <class Call>
+void
+expectThrowNamingKey(Call&& call)
+{
+    try
+    {
+        call();
+        ADD_FAILURE() << "expected SHAMapMissingNode, nothing was thrown";
+    }
+    catch (SHAMapMissingNode const& e)
+    {
+        std::string const what{e.what()};
+        EXPECT_NE(what.find(": id "), std::string::npos) << what;
+        EXPECT_EQ(what.find(": hash "), std::string::npos) << what;
+    }
+}
+
+/**
+ * A filter that resolves a fixed set of nodes, by hash.
+ *
+ * Stands in for the real sync filters, which serve a node from a local cache
+ * keyed on its hash, so the hash is all they check.
+ */
+class FixedNodeFilter : public SHAMapSyncFilter
+{
+    std::map<SHAMapHash, Blob> nodes_;
+
+public:
+    FixedNodeFilter(SHAMapHash const& hash, Blob blob)
+    {
+        nodes_.emplace(hash, std::move(blob));
+    }
+
+    explicit FixedNodeFilter(std::vector<std::pair<SHAMapHash, Blob>> nodes)
+    {
+        for (auto& [hash, blob] : nodes)
+            nodes_.emplace(hash, std::move(blob));
+    }
+
+    void
+    gotNode(
+        bool,
+        SHAMapHash const&,
+        std::uint32_t,
+        Blob&&,  // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved)
+        SHAMapNodeType) const override
+    {
+    }
+
+    [[nodiscard]] std::optional<Blob>
+    getNode(SHAMapHash const& hash) const override
+    {
+        if (auto const it = nodes_.find(hash); it != nodes_.end())
+            return it->second;
+        return std::nullopt;
+    }
+};
+
+// A tree whose hashes all agree can still put a leaf where its key does not belong, because a hash
+// covers a node's contents rather than its position.
+class SHAMapMisplacedLeaf : public ::testing::Test
+{
+protected:
+    beast::Journal const j_{TestSink::instance()};
+
+    // An arbitrary key whose first nibble is 1, so its leaf belongs under branch 1 of the root.
+    static constexpr UInt256 kKey{
+        "1c8cec8e5e9b0e5e0e0f5b3e2c9f7a1d6b4e8c2a0d7f3b9e5c1a8d4f2b6e0c93"};
+
+    // Any branch other than the one kKey selects at depth 0.
+    static constexpr unsigned int kWrongBranch = 5;
+
+    // The sequence a test writes stored nodes under, and sets on the map that reads them back. Any
+    // value serves: the nodestore is keyed by hash and takes this only as a lookup hint.
+    static constexpr std::uint32_t kStoredLedgerSeq = 11;
+
+    /**
+     * A genuine leaf holding kKey, in the form a sync filter serves, with its
+     * hash.
+     *
+     * Taken from a map that placed the leaf correctly, so only its position is
+     * ever wrong below. Serialized with its prefix rather than in wire form,
+     * since that is what checkFilter parses.
+     *
+     * @param f the family the throwaway source map belongs to.
+     * @return the leaf's prefixed form and its hash, or an empty blob if the
+     *         map rejected the item.
+     */
+    static std::pair<Blob, SHAMapHash>
+    genuineLeaf(Family& f)
+    {
+        SHAMap source{SHAMapType::FREE, f};
+        source.setUnbacked();
+        if (!source.addItem(
+                SHAMapNodeType::TnAccountState,
+                makeShamapitem(kKey, Slice{kKey.data(), kKey.size()})))
+        {
+            return {};
+        }
+
+        auto const path = source.getProofPath(kKey);
+        if (!path.has_value() || path->empty())
+            return {};
+
+        auto leaf = SHAMapTreeNode::makeFromWire(makeSlice(path->front()));
+        if (!leaf || !leaf->isLeaf())
+            return {};
+        leaf->updateHash();
+
+        Serializer s;
+        leaf->serializeWithPrefix(s);
+        return {s.getData(), leaf->getHash()};
+    }
+
+    /**
+     * Assemble `map` as a root inner node holding the given children.
+     *
+     * The root is installed directly, as a peer's would be, so each child stays
+     * unresolved until a walk consults the filter for it. Takes a list rather
+     * than one hash because a walk only reaches the sideways scans in
+     * peekNextItem and boundHelper from a branch it can already resolve, which
+     * needs a second, well-placed child.
+     *
+     * @param map the map to assemble, which must be synching and empty.
+     * @param children the branch and hash of each child the forged root
+     *        records.
+     * @return whether the root was accepted.
+     */
+    static bool
+    forgeRoot(SHAMap& map, std::vector<InnerChild> const& children)
+    {
+        auto root = makeFullInnerNode(children);
+        if (!root)
+            return false;
+
+        auto const rootHash = root->getHash();
+        return map.addRootNode(rootHash, std::move(root), nullptr).isGood();
+    }
+
+    /**
+     * A correctly-built leaf for the given key, in the form a sync filter
+     * serves.
+     *
+     * Serialized with its prefix rather than in wire form, since that is what
+     * checkFilter parses.
+     *
+     * @param f the family the source map belongs to.
+     * @param key the key to build the leaf for.
+     * @return the leaf's prefixed form and its hash, or an empty blob if the
+     *         map rejected the item.
+     */
+    static std::pair<Blob, SHAMapHash>
+    leafFor(Family& f, UInt256 const& key)
+    {
+        SHAMap source{SHAMapType::FREE, f};
+        source.setUnbacked();
+        if (!source.addItem(
+                SHAMapNodeType::TnAccountState, makeShamapitem(key, Slice{key.data(), key.size()})))
+        {
+            return {};
+        }
+
+        auto const path = source.getProofPath(key);
+        if (!path.has_value() || path->empty())
+            return {};
+
+        // NOLINTNEXTLINE(bugprone-unchecked-optional-access) has_value() checked above
+        auto const leaf = SHAMapTreeNode::makeFromWire(makeSlice(path->front()));
+        if (!leaf || !leaf->isLeaf())
+            return {};
+
+        Serializer s;
+        leaf->serializeWithPrefix(s);
+        return {s.getData(), leaf->getHash()};
+    }
+
+    /**
+     * A key whose first nibble equals its last.
+     *
+     * Any key builds a chain. This one also lets a refusal be told apart from
+     * an accident: a walk that ignored the depth bound selects, at the level
+     * past kLeafDepth, the one branch the node there occupies, and so reports
+     * a child it could not fetch instead of answering.
+     */
+    static constexpr UInt256 kChainKeyPastLeafDepth{
+        "b92891fe4ef6cee585fdc6fda1e09eb4d386363158ec3321b8123e5a772c6cab"};
+
+    /**
+     * Store a chain of single-child inner nodes running one level past
+     * kLeafDepth.
+     *
+     * Built bottom-up, so each parent records the hash of the node below it and
+     * the chain verifies at every level. The deepest node points at a child the
+     * store does not hold.
+     *
+     * @param f the family whose store the chain is written to.
+     * @param key the key whose nibbles choose the branch at every level.
+     * @return the chain's root and the hash of the node at kLeafDepth, or an
+     *         empty pair if a node could not be built.
+     */
+    static std::pair<SHAMapTreeNodePtr, SHAMapHash>
+    storeInnerChainPastLeafDepth(tests::TestNodeFamily& f, UInt256 const& key)
+    {
+        SHAMapHash childHash{UInt256{1}};
+        SHAMapTreeNodePtr node;
+        SHAMapHash deepestHash;
+
+        // The decrement is in the body, so the counter never steps below zero.
+        // UndefinedBehaviorSanitizer reports that wraparound.
+        for (auto depth = SHAMap::kLeafDepth + 1u; depth > 0;)
+        {
+            --depth;
+            auto const id = SHAMapNodeID::createID(std::min(depth, SHAMap::kLeafDepth - 1u), key);
+            node = makeFullInnerNode({{.branch = selectBranch(id, key), .hash = childHash}});
+            if (!node)
+                return {};
+
+            Serializer prefixed;
+            node->serializeWithPrefix(prefixed);
+            f.db().store(
+                NodeObjectType::AccountNode,
+                prefixed.getData(),
+                node->getHash().asUInt256(),
+                kStoredLedgerSeq);
+
+            childHash = node->getHash();
+            if (depth == SHAMap::kLeafDepth)
+                deepestHash = childHash;
+        }
+
+        return {std::move(node), deepestHash};
+    }
+};
+
+// Every branch above a leaf is judged, not only the last one, so a leaf under a subtree recorded on
+// a branch its key does not select is refused.
+TEST_F(SHAMapMisplacedLeaf, iterating_a_misplaced_subtree_throws)
+{
+    // A key greater than the probe below, so a greater key exists for that probe to find.
+    constexpr UInt256 kDeepKey{"a100000000000000000000000000000000000000000000000000000000000000"};
+
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [leafBlob, leafHash] = leafFor(sourceFamily, kDeepKey);
+    ASSERT_FALSE(leafBlob.empty());
+
+    // The subtree holds exactly one child, on the branch its key selects at depth 1, so the walk
+    // resolves the same node whichever branch the random scan starts from.
+    auto const leafBranch = selectBranch(1u, kDeepKey);
+    auto const inner = makeFullInnerNode({{.branch = leafBranch, .hash = leafHash}});
+    ASSERT_TRUE(inner);
+
+    Serializer innerPrefixed;
+    inner->serializeWithPrefix(innerPrefixed);
+
+    // The leaf is served as well, so the walk reaches it and the throw below is about position.
+    std::vector<std::pair<SHAMapHash, Blob>> served;
+    served.emplace_back(inner->getHash(), innerPrefixed.getData());
+    served.emplace_back(leafHash, leafBlob);
+
+    tests::TestNodeFamily targetFamily{j_};
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setUnbacked();
+    ASSERT_TRUE(forgeRoot(map, {{.branch = kWrongBranch, .hash = inner->getHash()}}));
+
+    // The inner node carries no key, so the leaf below it is what shows the branch it was reached
+    // through disagrees with the key underneath.
+    FixedNodeFilter const filter{std::move(served)};
+    map.getMissingNodes(4, &filter);
+
+    EXPECT_THROW(map.begin(), SHAMapMissingNode);
+
+    // The bounds refuse the same map, and a refusal throws where end() answers. The probe selects
+    // kWrongBranch at depth 0 and leafBranch at depth 1, so the descent walks straight into the
+    // misplaced leaf and clears the path. The one key in the map is greater than the probe.
+    UInt256 probe;
+    probe.begin()[0] = static_cast<std::uint8_t>((kWrongBranch << 4) | leafBranch);
+    ASSERT_GT(kDeepKey, probe);
+
+    EXPECT_THROW(map.upperBound(probe), SHAMapMissingNode);
+    EXPECT_THROW(map.lowerBound(probe), SHAMapMissingNode);
+}
+
+// peekNextItem scans the branches beyond the one a key takes and judges each node it resolves
+// there. The root holds a well-placed leaf so the scan starts, and a misplaced one for it to reach.
+TEST_F(SHAMapMisplacedLeaf, iterating_past_a_leaf_into_a_misplaced_sibling_throws)
+{
+    // First nibble 10, so this leaf sits where it belongs. kKey's first nibble is 1, so its leaf
+    // does not belong on kStrayBranch, which is above 10 and so within the forward scan's reach.
+    constexpr UInt256 kPlacedKey{
+        "a000000000000000000000000000000000000000000000000000000000000000"};
+    static constexpr unsigned int kStrayBranch = 12;
+
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [placedBlob, placedHash] = leafFor(sourceFamily, kPlacedKey);
+    ASSERT_FALSE(placedBlob.empty());
+    auto const [strayBlob, strayHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(strayBlob.empty());
+
+    // Both leaves go in the store rather than a filter, because the scan reaches them through
+    // descendThrow(), which consults no filter. Served from the store, the throw is about
+    // position.
+    tests::TestNodeFamily targetFamily{j_};
+    targetFamily.db().store(
+        NodeObjectType::AccountNode, Blob{placedBlob}, placedHash.asUInt256(), kStoredLedgerSeq);
+    targetFamily.db().store(
+        NodeObjectType::AccountNode, Blob{strayBlob}, strayHash.asUInt256(), kStoredLedgerSeq);
+
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setLedgerSeq(kStoredLedgerSeq);
+
+    auto const placedBranch = selectBranch(SHAMapNodeID{}, kPlacedKey);
+    ASSERT_LT(placedBranch, kStrayBranch);
+    ASSERT_TRUE(forgeRoot(
+        map,
+        {{.branch = placedBranch, .hash = placedHash},
+         {.branch = kStrayBranch, .hash = strayHash}}));
+
+    // The store really holds both, so a refusal below is about position.
+    ASSERT_NE(targetFamily.db().fetchNodeObject(placedHash.asUInt256(), kStoredLedgerSeq), nullptr);
+    ASSERT_NE(targetFamily.db().fetchNodeObject(strayHash.asUInt256(), kStoredLedgerSeq), nullptr);
+
+    // The well-placed leaf resolves, so iteration starts.
+    auto it = map.begin();
+    ASSERT_NE(it, map.end());
+    EXPECT_EQ(it->key(), kPlacedKey);
+
+    // Advancing scans past placedBranch, resolves the stray leaf out of the store, and pushChild
+    // refuses it. The payload check narrows this to peekNextItem's two id-form throws, of which
+    // only the refused push can occur against this fixture.
+    expectThrowNamingKey([&] { ++it; });
+}
+
+// boundHelper runs the same sideways scan as peekNextItem and has its own throw for a node it
+// cannot push. Reached the same way, from a probe whose own branch resolves.
+TEST_F(SHAMapMisplacedLeaf, a_bound_scanning_into_a_misplaced_sibling_throws)
+{
+    constexpr UInt256 kPlacedKey{
+        "a000000000000000000000000000000000000000000000000000000000000000"};
+    static constexpr unsigned int kStrayBranch = 12;
+
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [placedBlob, placedHash] = leafFor(sourceFamily, kPlacedKey);
+    ASSERT_FALSE(placedBlob.empty());
+    auto const [strayBlob, strayHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(strayBlob.empty());
+
+    tests::TestNodeFamily targetFamily{j_};
+    targetFamily.db().store(
+        NodeObjectType::AccountNode, Blob{placedBlob}, placedHash.asUInt256(), kStoredLedgerSeq);
+    targetFamily.db().store(
+        NodeObjectType::AccountNode, Blob{strayBlob}, strayHash.asUInt256(), kStoredLedgerSeq);
+
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setLedgerSeq(kStoredLedgerSeq);
+
+    auto const placedBranch = selectBranch(SHAMapNodeID{}, kPlacedKey);
+    ASSERT_LT(placedBranch, kStrayBranch);
+    ASSERT_TRUE(forgeRoot(
+        map,
+        {{.branch = placedBranch, .hash = placedHash},
+         {.branch = kStrayBranch, .hash = strayHash}}));
+
+    ASSERT_NE(targetFamily.db().fetchNodeObject(placedHash.asUInt256(), kStoredLedgerSeq), nullptr);
+    ASSERT_NE(targetFamily.db().fetchNodeObject(strayHash.asUInt256(), kStoredLedgerSeq), nullptr);
+
+    // upperBound from the well-placed key descends to its leaf, then scans above it and reaches the
+    // misplaced one, whose own key is greater than the probe. Narrowed to boundHelper's two id-form
+    // throws, of which only the refused push can occur against this fixture.
+    expectThrowNamingKey([&] { static_cast<void>(map.upperBound(kPlacedKey)); });
+}
+
+// walkTowardsKey refuses a node past kLeafDepth in both its modes. The mode that records a path for
+// its caller refuses through pushChild and clears the path, so the caller reads an empty path as a
+// refusal.
+TEST_F(SHAMapMisplacedLeaf, a_keyed_walk_recording_a_path_refuses_a_node_past_leaf_depth)
+{
+    tests::TestNodeFamily targetFamily{j_};
+    auto [root, deepestHash] = storeInnerChainPastLeafDepth(targetFamily, kChainKeyPastLeafDepth);
+    ASSERT_TRUE(root);
+    ASSERT_TRUE(deepestHash.isNonZero());
+
+    // The node the walk has to refuse is really in the store, so the refusal is about depth.
+    ASSERT_NE(
+        targetFamily.db().fetchNodeObject(deepestHash.asUInt256(), kStoredLedgerSeq), nullptr);
+
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setLedgerSeq(kStoredLedgerSeq);
+    auto const rootHash = root->getHash();
+    ASSERT_TRUE(map.addRootNode(rootHash, std::move(root), nullptr).isGood());
+
+    // delItem hands walkTowardsKey a path to fill, reads the cleared path as a refusal, and names
+    // the key it was walking to.
+    expectThrowNamingKey([&] { static_cast<void>(map.delItem(kChainKeyPastLeafDepth)); });
+}
+
+// The other mode: a walk with no path at all tracks a depth of its own and refuses at the same
+// bound, answering rather than throwing.
+TEST_F(SHAMapMisplacedLeaf, a_keyed_walk_with_no_path_refuses_a_node_past_leaf_depth)
+{
+    tests::TestNodeFamily targetFamily{j_};
+    auto [root, deepestHash] = storeInnerChainPastLeafDepth(targetFamily, kChainKeyPastLeafDepth);
+    ASSERT_TRUE(root);
+    ASSERT_TRUE(deepestHash.isNonZero());
+
+    ASSERT_NE(
+        targetFamily.db().fetchNodeObject(deepestHash.asUInt256(), kStoredLedgerSeq), nullptr);
+
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setLedgerSeq(kStoredLedgerSeq);
+    auto const rootHash = root->getHash();
+    ASSERT_TRUE(map.addRootNode(rootHash, std::move(root), nullptr).isGood());
+
+    // hasItem reaches findKey, which walks with no path. The walk reports the refusal and stops, so
+    // the key reads as absent.
+    bool found = true;
+    ASSERT_NO_THROW(found = map.hasItem(kChainKeyPastLeafDepth));
+    EXPECT_FALSE(found);
+}
+
+// A path of inner nodes running past kLeafDepth has no room for the leaf it leads to, so the
+// traversal refuses it. The chain is resolvable, so the throw is about depth.
+TEST_F(SHAMapMisplacedLeaf, a_path_of_inner_nodes_past_leaf_depth_is_refused)
+{
+    // An arbitrary key. Only the branches it selects matter here.
+    constexpr UInt256 kChainKey{"b92891fe4ef6cee585fdc6fda1e09eb4d386363158ec3321b8123e5a772c6ca8"};
+
+    tests::TestNodeFamily targetFamily{j_};
+
+    // Built bottom-up, so each parent records the hash of the node below it and the chain verifies
+    // at every level. The deepest node points at a child the store does not hold.
+    SHAMapHash childHash{UInt256{1}};
+    SHAMapTreeNodePtr node;
+
+    // The node at kLeafDepth, which is the one the traversal has to refuse. Recorded so the refusal
+    // can be told apart from a failure to fetch the sentinel the chain ends on.
+    SHAMapHash deepestHash;
+
+    // The decrement is in the body, so the counter never steps below zero.
+    // UndefinedBehaviorSanitizer reports that wraparound.
+    for (auto depth = SHAMap::kLeafDepth + 1u; depth > 0;)
+    {
+        --depth;
+        auto const id = SHAMapNodeID::createID(std::min(depth, SHAMap::kLeafDepth - 1u), kChainKey);
+        node = makeFullInnerNode({{.branch = selectBranch(id, kChainKey), .hash = childHash}});
+        ASSERT_TRUE(node);
+
+        Serializer prefixed;
+        node->serializeWithPrefix(prefixed);
+        targetFamily.db().store(
+            NodeObjectType::AccountNode,
+            prefixed.getData(),
+            node->getHash().asUInt256(),
+            kStoredLedgerSeq);
+
+        childHash = node->getHash();
+        if (depth == SHAMap::kLeafDepth)
+            deepestHash = childHash;
+    }
+    ASSERT_TRUE(deepestHash.isNonZero());
+
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setLedgerSeq(kStoredLedgerSeq);
+    ASSERT_TRUE(map.addRootNode(childHash, std::move(node), nullptr).isGood());
+
+    // The chain is really readable, so the refusal below is about depth.
+    ASSERT_NE(targetFamily.db().fetchNodeObject(childHash.asUInt256(), kStoredLedgerSeq), nullptr);
+
+    // Refused with SHAMapMissingNode, which is pushChild's other refusal shape: too deep rather
+    // than misplaced. The depth bound applies before getChildNodeID is asked for a child.
+    try
+    {
+        static_cast<void>(map.begin());
+        ADD_FAILURE() << "the traversal did not refuse the path";
+    }
+    catch (SHAMapMissingNode const& e)
+    {
+        // belowHelper's refusal reports the hash of the child it refused, where the scan sites
+        // above report the key they were searching for. So the node named here is the one at
+        // kLeafDepth.
+        std::string const what{e.what()};
+        EXPECT_NE(what.find(to_string(deepestHash)), std::string::npos) << what;
+    }
+}
+
+// addKnownNode reaches a filter through the synchronous descend on its way to the position it was
+// given, which is the other route a node takes into a tree during acquisition.
+//
+// What this pins is the verdict addKnownNode reports and that the map ends invalid.
+TEST_F(SHAMapMisplacedLeaf, hooking_a_known_node_invalidates_the_map)
+{
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [leafBlob, leafHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(leafBlob.empty());
+
+    tests::TestNodeFamily targetFamily{j_};
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setUnbacked();
+    ASSERT_TRUE(forgeRoot(map, {{.branch = kWrongBranch, .hash = leafHash}}));
+    ASSERT_TRUE(map.isValid());
+
+    // A key whose first nibble is kWrongBranch, so the walk descends the branch holding the leaf.
+    // An inner node is offered rather than a leaf, so the walk reaches the descent and judges
+    // what it resolves there.
+    auto const target = SHAMapNodeID::createID(
+        2, UInt256{"5000000000000000000000000000000000000000000000000000000000000000"});
+
+    Serializer s;
+    for (auto i = 0u; i < SHAMap::kBranchFactor; ++i)
+        s.addBitString(i == 0u ? UInt256{1} : UInt256{});
+    s.add8(kWireTypeInner);
+    auto offered = SHAMapTreeNode::makeFromWire(makeSlice(s.peekData()));
+    ASSERT_TRUE(offered);
+    offered->updateHash();
+
+    FixedNodeFilter const filter{leafHash, leafBlob};
+    auto const result = map.addKnownNode(target, std::move(offered), &filter);
+
+    EXPECT_FALSE(map.isValid());
+
+    // The verdict matters as much as the state: it is what the acquisition paths charge a peer on,
+    // so a later change to it should fail here rather than pass quietly.
+    EXPECT_TRUE(result.isInvalid());
+    EXPECT_FALSE(result.isGood());
+}
+
+// The filter descent judges depth as well as position, and this case reaches the depth arm. It
+// serves an inner node, for which belongsAt holds by definition, so the verdict below can only come
+// from pastLeafDepth. Removing that conjunct from descend() makes this case fail.
+//
+// The whole chain is served through the filter and nothing is written to the store, because the
+// memory nodestore is shared by path across every test family in this binary.
+TEST_F(SHAMapMisplacedLeaf, a_filter_serving_an_inner_node_at_leaf_depth_invalidates_the_map)
+{
+    // A key of this case's own, so the chain it builds is its own too.
+    constexpr UInt256 kOwnChainKey{
+        "3f0e1d2c3b4a59687766554433221100ffeeddccbbaa99887766554433221100"};
+
+    // Built bottom-up, so each parent records the hash of the node below it. Every node goes to the
+    // filter, including the one at kLeafDepth, which is the one the descent has to refuse.
+    std::vector<std::pair<SHAMapHash, Blob>> served;
+    SHAMapHash childHash{UInt256{1}};
+    SHAMapTreeNodePtr node;
+
+    // The decrement is in the body, so the counter never steps below zero.
+    // UndefinedBehaviorSanitizer reports that wraparound.
+    for (auto depth = SHAMap::kLeafDepth + 1u; depth > 0;)
+    {
+        --depth;
+        auto const id =
+            SHAMapNodeID::createID(std::min(depth, SHAMap::kLeafDepth - 1u), kOwnChainKey);
+        node = makeFullInnerNode({{.branch = selectBranch(id, kOwnChainKey), .hash = childHash}});
+        ASSERT_TRUE(node);
+        ASSERT_TRUE(node->isInner());
+
+        Serializer prefixed;
+        node->serializeWithPrefix(prefixed);
+        served.emplace_back(node->getHash(), prefixed.getData());
+
+        childHash = node->getHash();
+    }
+    ASSERT_EQ(served.size(), SHAMap::kLeafDepth + 1u);
+
+    tests::TestNodeFamily targetFamily{j_};
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setUnbacked();
+    ASSERT_TRUE(map.addRootNode(childHash, std::move(node), nullptr).isGood());
+    ASSERT_TRUE(map.isValid());
+
+    // kLeafDepth is the deepest ID the tree allows, so the walk descends from kLeafDepth - 1 once
+    // more and consults the filter for a child at kLeafDepth.
+    auto const target = SHAMapNodeID::createID(SHAMap::kLeafDepth, kOwnChainKey);
+
+    Serializer offeredWire;
+    for (auto i = 0u; i < SHAMap::kBranchFactor; ++i)
+        offeredWire.addBitString(i == 0u ? UInt256{1} : UInt256{});
+    offeredWire.add8(kWireTypeInner);
+    auto offered = SHAMapTreeNode::makeFromWire(makeSlice(offeredWire.peekData()));
+    ASSERT_TRUE(offered);
+    offered->updateHash();
+
+    FixedNodeFilter const filter{std::move(served)};
+    static_cast<void>(map.addKnownNode(target, std::move(offered), &filter));
+
+    EXPECT_FALSE(map.isValid());
+}
+
+// addKnownNode also hooks the very node it was handed, on the path where the local store has
+// nothing to resolve for that slot. Such a node's position is known only from the ID the caller
+// supplied, so it is judged against the leaf's own key before it is hooked.
+TEST_F(SHAMapMisplacedLeaf, hooking_an_offered_misplaced_leaf_invalidates_the_map)
+{
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [leafBlob, leafHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(leafBlob.empty());
+
+    tests::TestNodeFamily targetFamily{j_};
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setUnbacked();
+    ASSERT_TRUE(forgeRoot(map, {{.branch = kWrongBranch, .hash = leafHash}}));
+    ASSERT_TRUE(map.isValid());
+
+    // The branch the forged root files the leaf under, which is not the one kKey selects.
+    UInt256 wrongPrefix;
+    wrongPrefix.begin()[0] = static_cast<std::uint8_t>(kWrongBranch << 4);
+    auto const target = SHAMapNodeID::createID(1, wrongPrefix);
+
+    auto offered = SHAMapTreeNode::makeFromPrefix(makeSlice(leafBlob), leafHash);
+    ASSERT_TRUE(offered);
+    ASSERT_TRUE(offered->isLeaf());
+
+    // No filter, so the walk resolves nothing locally and the node offered here is the one hooked.
+    auto const result = map.addKnownNode(target, std::move(offered), nullptr);
+
+    EXPECT_FALSE(map.isValid());
+    EXPECT_TRUE(result.isInvalid());
+    EXPECT_FALSE(result.isGood());
+
+    // The verdict names which arm refused, not just that one did. This arm condemned the map, so
+    // it reports mapInvalidated(); the arm that only rejects a misplaced label reports the plain
+    // invalid() and leaves the map usable, which
+    // SHAMapSyncTest.only_the_map_invalidating_arm_reports_it holds the other side of. The two
+    // decide which fee InboundLedger::receiveNode() charges, so the names cannot be swapped.
+    EXPECT_TRUE(result.invalidatedMap());
+}
+
+// getMissingNodes reaches a filter through descendAsync, which hooks whatever it resolves. The
+// verdict lands on the map, since every node from the root down hash-verified to get here.
+TEST_F(SHAMapMisplacedLeaf, walking_for_missing_nodes_invalidates_the_map)
+{
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [leafBlob, leafHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(leafBlob.empty());
+
+    // Its own family, so the leaf is reachable only through the filter rather than from a cache the
+    // source map warmed.
+    tests::TestNodeFamily targetFamily{j_};
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setUnbacked();
+    ASSERT_TRUE(forgeRoot(map, {{.branch = kWrongBranch, .hash = leafHash}}));
+    ASSERT_TRUE(map.isValid());
+
+    FixedNodeFilter const filter{leafHash, leafBlob};
+    map.getMissingNodes(1, &filter);
+
+    EXPECT_FALSE(map.isValid());
+}
+
+// The descendAsync walk leaves the leaf hooked, since it resolves the node before the position
+// is judged. Iterating it throws SHAMapMissingNode, which is how belowHelper reports a refusal.
+TEST_F(SHAMapMisplacedLeaf, iterating_a_hooked_misplaced_leaf_throws)
+{
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [leafBlob, leafHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(leafBlob.empty());
+
+    tests::TestNodeFamily targetFamily{j_};
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setUnbacked();
+    ASSERT_TRUE(forgeRoot(map, {{.branch = kWrongBranch, .hash = leafHash}}));
+
+    FixedNodeFilter const filter{leafHash, leafBlob};
+    map.getMissingNodes(1, &filter);
+    ASSERT_FALSE(map.isValid());
+
+    EXPECT_THROW(map.begin(), SHAMapMissingNode);
+}
+
+// The walk that resolves a node through descendAsync is the production path, since a backed map
+// posts asynchronous reads rather than fetching inline. This case is the one that reaches that
+// guard, since it uses a backed map with the leaf in the store.
+TEST_F(SHAMapMisplacedLeaf, an_async_read_resolving_a_misplaced_leaf_invalidates_the_map)
+{
+    tests::TestNodeFamily sourceFamily{j_};
+    auto const [leafBlob, leafHash] = genuineLeaf(sourceFamily);
+    ASSERT_FALSE(leafBlob.empty());
+
+    // Backed and with the leaf in the store, so asyncFetch resolves it. No filter is passed, which
+    // is what forces the walk down the asynchronous route rather than through checkFilter.
+    tests::TestNodeFamily targetFamily{j_};
+    targetFamily.db().store(
+        NodeObjectType::AccountNode, Blob{leafBlob}, leafHash.asUInt256(), kStoredLedgerSeq);
+
+    SHAMap map{SHAMapType::FREE, UInt256{}, targetFamily};
+    map.setLedgerSeq(kStoredLedgerSeq);
+    ASSERT_TRUE(forgeRoot(map, {{.branch = kWrongBranch, .hash = leafHash}}));
+
+    map.getMissingNodes(1, nullptr);
+
+    EXPECT_FALSE(map.isValid());
+}
 }  // namespace xrpl::tests

src/tests/libxrpl/shamap/SHAMapAddNode.cpp

@@ -0,0 +1,76 @@
+#include <xrpl/shamap/SHAMapAddNode.h>
+
+#include <gtest/gtest.h>
+
+namespace xrpl::tests {
+
+// get() is a log format, so its wording is pinned here, once. Every other site reads the same tally
+// by value, through the count and verdict accessors.
+TEST(SHAMapAddNode, get_names_every_non_empty_count)
+{
+    EXPECT_EQ(SHAMapAddNode{}.get(), "no nodes processed");
+    EXPECT_EQ(SHAMapAddNode::useful().get(), "good:1");
+    EXPECT_EQ(SHAMapAddNode::invalid().get(), "bad:1");
+    EXPECT_EQ(SHAMapAddNode::duplicate().get(), "dupe:1");
+
+    // Several of a kind are counted, and the counts are joined in a fixed order with a single
+    // space, whichever order they were recorded in.
+    SHAMapAddNode san;
+    san.incInvalid();
+    san.incUseful();
+    san.incUseful();
+    san.incDuplicate();
+    EXPECT_EQ(san.get(), "good:2 bad:1 dupe:1");
+
+    san.reset();
+    EXPECT_EQ(san.get(), "no nodes processed");
+}
+
+// The three counts and the verdicts derived from them.
+TEST(SHAMapAddNode, counts_and_verdicts_agree)
+{
+    SHAMapAddNode san;
+    EXPECT_EQ(san.getGood(), 0);
+    EXPECT_EQ(san.getBad(), 0);
+    EXPECT_EQ(san.getDuplicate(), 0);
+    EXPECT_FALSE(san.isInvalid());
+    EXPECT_FALSE(san.isUseful());
+
+    // Good counts what produced a good result, and useful is that count being non-zero.
+    san.incUseful();
+    EXPECT_EQ(san.getGood(), 1);
+    EXPECT_TRUE(san.isUseful());
+    EXPECT_TRUE(san.isGood());
+
+    // A duplicate counts toward good. isUseful() here reflects the incUseful() above.
+    san.incDuplicate();
+    EXPECT_EQ(san.getDuplicate(), 1);
+    EXPECT_FALSE(san.isInvalid());
+    EXPECT_TRUE(san.isGood());
+
+    // Bad is a count, so a batch that carries on past a rejected node reports one per node, which
+    // distinguishes "stopped on the first" from "rejected several".
+    san.incInvalid();
+    EXPECT_EQ(san.getBad(), 1);
+    EXPECT_TRUE(san.isInvalid());
+    EXPECT_TRUE(san.isGood()) << "one bad node among two accepted ones is still a good batch";
+
+    san.incInvalid();
+    san.incInvalid();
+    EXPECT_EQ(san.getGood(), 1);
+    EXPECT_EQ(san.getBad(), 3);
+    EXPECT_EQ(san.getDuplicate(), 1);
+    EXPECT_FALSE(san.isGood()) << "more bad nodes than accepted ones is not a good batch";
+
+    // Adding one verdict to another sums every count.
+    SHAMapAddNode total;
+    total += SHAMapAddNode::useful();
+    total += SHAMapAddNode::invalid();
+    total += SHAMapAddNode::invalid();
+    total += SHAMapAddNode::duplicate();
+    EXPECT_EQ(total.getGood(), 1);
+    EXPECT_EQ(total.getBad(), 2);
+    EXPECT_EQ(total.getDuplicate(), 1);
+}
+
+}  // namespace xrpl::tests

src/tests/libxrpl/shamap/SHAMapNodeID.cpp

@@ -90,6 +90,33 @@ TEST(SHAMapNodeIDTest, create_id_masks_key_to_depth)
     }
 }
 
+// A child's id is its parent's with the nibble at the parent's own depth set to the branch taken.
+// The reference comes from createID, so it is produced by the depthMask table rather than by the
+// arithmetic under test.
+TEST(SHAMapNodeIDTest, child_node_id_sets_the_nibble_the_next_depth_masks_in)
+{
+    for (auto depth = 0u; depth < SHAMap::kLeafDepth; ++depth)
+    {
+        auto const parent = SHAMapNodeID::createID(depth, kTestKey);
+
+        // The branch kTestKey's own nibble selects at this depth, so the child sits where createID
+        // puts that key one level down.
+        auto const branch = selectBranch(parent, kTestKey);
+        auto const child = SHAMapNodeID::createID(depth + 1, kTestKey);
+        auto const& expected = child.getNodeID();
+
+        EXPECT_EQ(childNodeID(parent.getNodeID(), depth, branch), expected) << "depth " << depth;
+
+        // Only that branch reaches that id, so the whole branch value lands in the nibble and
+        // sixteen branches name sixteen distinct positions.
+        for (auto other = 0u; other < SHAMap::kBranchFactor; ++other)
+        {
+            EXPECT_EQ(childNodeID(parent.getNodeID(), depth, other) == expected, other == branch)
+                << "depth " << depth << " branch " << other;
+        }
+    }
+}
+
 // The guards below must hold with XRPL_ASSERT compiled out (NDEBUG), so each one
 // has to be a real runtime check rather than an assert.
 
@@ -165,6 +192,43 @@ TEST(SHAMapNodeIDDeathTest, select_branch_clamps_leaf_depth)
 #endif
 }
 
+TEST(SHAMapNodeIDDeathTest, same_position_at_depth_holds_its_own_bound)
+{
+    // Differs from kTestKey in the last nibble only, so the two agree at every depth below
+    // kLeafDepth and only kLeafDepth itself tells them apart.
+    constexpr UInt256 kLastNibbleDiffers(
+        "b92891fe4ef6cee585fdc6fda1e09eb4d386363158ec3321b8123e5a772c6ca9");
+
+    // Depth 0 names no nibble, so every key shares the root position.
+    EXPECT_TRUE(samePositionAtDepth(0, kTestKey, kLastNibbleDiffers));
+    EXPECT_TRUE(samePositionAtDepth(0, kTestKey, UInt256{}));
+
+    // kLeafDepth is in range here, unlike in selectBranch, since the position at that depth is
+    // the whole key. A bound borrowed from selectBranch would clamp this to 63 and call the two
+    // keys equal, so the pair below is what holds the bound at the right value.
+    EXPECT_TRUE(samePositionAtDepth(SHAMap::kLeafDepth - 1u, kTestKey, kLastNibbleDiffers));
+    EXPECT_FALSE(samePositionAtDepth(SHAMap::kLeafDepth, kTestKey, kLastNibbleDiffers));
+    EXPECT_TRUE(samePositionAtDepth(SHAMap::kLeafDepth, kTestKey, kTestKey));
+
+    // Past kLeafDepth there is no mask in depthMask's 65-entry table, so the call clamps. That
+    // clamp is marked UNREACHABLE, which is an assert and therefore fatal wherever asserts are
+    // live, so only a build with them compiled out (or routed to Antithesis's non-fatal handler)
+    // reaches the clamp itself and can be asserted on.
+#if defined(NDEBUG) || defined(ENABLE_VOIDSTAR)
+    for (auto const depth : {SHAMap::kLeafDepth + 1u, 100u, 255u, 256u, 320u})
+    {
+        // Answers as kLeafDepth does, rather than reading past the table or narrowing the depth
+        // to a byte: 256 would otherwise become 0 and call the two keys equal.
+        EXPECT_FALSE(samePositionAtDepth(depth, kTestKey, kLastNibbleDiffers)) << "depth " << depth;
+        EXPECT_TRUE(samePositionAtDepth(depth, kTestKey, kTestKey)) << "depth " << depth;
+    }
+#else
+    EXPECT_DEATH(
+        (void)samePositionAtDepth(SHAMap::kLeafDepth + 1u, kTestKey, kTestKey),
+        "depth within tree");
+#endif
+}
+
 TEST(SHAMapNodeIDTest, deserialize_rejects_out_of_range_depth)
 {
     // getRawString() only serializes a depth already accepted by the constructor's own

src/tests/libxrpl/shamap/common.h

@@ -14,7 +14,9 @@
 #include <xrpl/shamap/FullBelowCache.h>
 #include <xrpl/shamap/TreeNodeCache.h>
 
+#include <atomic>
 #include <chrono>
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <stdexcept>
@@ -26,16 +28,27 @@ class TestNodeFamily : public Family
 private:
     std::unique_ptr<node_store::Database> db_;
 
+    // Declared before the two caches, which both bind a reference to it in the initializer list.
+    TestStopwatch clock_;
+
     std::shared_ptr<FullBelowCache> fbCache_;
     std::shared_ptr<TreeNodeCache> tnCache_;
 
-    TestStopwatch clock_;
     node_store::DummyScheduler scheduler_;
 
     beast::Journal const j_;
 
+    // Written from whichever nodestore reader thread reports the miss, so read back atomically.
+    std::atomic<std::size_t> missingBySeqReports_ = 0;
+    std::atomic<std::uint32_t> missingBySeqRefNum_ = 0;
+
 public:
-    TestNodeFamily(beast::Journal j)
+    /**
+     * @param j The journal to log through.
+     * @param readThreads How many nodestore reader threads to run asynchronous
+     *        fetches on.
+     */
+    explicit TestNodeFamily(beast::Journal j, int readThreads = 1)
         : fbCache_(std::make_shared<FullBelowCache>("App family full below cache", clock_, j))
         , tnCache_(
               std::make_shared<TreeNodeCache>(
@@ -50,7 +63,7 @@ public:
         testSection.set(Keys::kType, "memory");
         testSection.set(Keys::kPath, "SHAMap_test");
         db_ = node_store::Manager::instance().makeDatabase(
-            megabytes(4), scheduler_, 1, testSection, j);
+            megabytes(4), scheduler_, readThreads, testSection, j);
     }
 
     node_store::Database&
@@ -90,14 +103,28 @@ public:
         tnCache_->sweep();
     }
 
+    /**
+     * Record the report and throw, standing in for Family's real acquisition
+     * machinery.
+     *
+     * @param refNum Sequence of the ledger with the missing node. Recorded, and
+     *        readable through missingBySeqRefNum().
+     * @param nodeHash Hash of the missing node. Unused.
+     */
     void
-    missingNodeAcquireBySeq(
-        [[maybe_unused]] std::uint32_t refNum,
-        [[maybe_unused]] UInt256 const& nodeHash) override
+    missingNodeAcquireBySeq(std::uint32_t refNum, [[maybe_unused]] UInt256 const& nodeHash) override
     {
+        missingBySeqRefNum_.store(refNum, std::memory_order_release);
+        ++missingBySeqReports_;
         Throw<std::runtime_error>("missing node");
     }
 
+    /**
+     * Throw, standing in for Family's real acquisition machinery. Uncounted.
+     *
+     * @param refHash Hash of the ledger with the missing node. Unused.
+     * @param refNum Sequence of the ledger with the missing node. Unused.
+     */
     void
     missingNodeAcquireByHash(
         [[maybe_unused]] UInt256 const& refHash,
@@ -106,6 +133,29 @@ public:
         Throw<std::runtime_error>("missing node");
     }
 
+    /**
+     * How many times a map of this family has withdrawn its claim of being
+     * complete in the database. Counted per family, not per map.
+     *
+     * @return The number of missingNodeAcquireBySeq() calls so far.
+     */
+    [[nodiscard]] std::size_t
+    missingBySeqReports() const
+    {
+        return missingBySeqReports_.load(std::memory_order_acquire);
+    }
+
+    /**
+     * The ledger sequence the most recent such report named.
+     *
+     * @return The sequence, or zero if nothing has been reported yet.
+     */
+    [[nodiscard]] std::uint32_t
+    missingBySeqRefNum() const
+    {
+        return missingBySeqRefNum_.load(std::memory_order_acquire);
+    }
+
     void
     reset() override
     {

src/xrpld/app/ledger/ConsensusTransSetSF.cpp

@@ -42,7 +42,7 @@ ConsensusTransSetSF::gotNode(
 
     nodeCache_.insert(nodeHash, nodeData);
 
-    if ((type == SHAMapNodeType::TnTransactionNm) && (nodeData.size() > 16))
+    if ((type == SHAMapNodeType::TnTransactionNm) && (nodeData.size() >= kMinTxNodeBytesToParse))
     {
         // this is a transaction, and we didn't have it
         JLOG(j_.debug()) << "Node on our acquiring TX set is TXN we may not have";

src/xrpld/app/ledger/ConsensusTransSetSF.h

@@ -9,6 +9,7 @@
 #include <xrpl/shamap/SHAMapSyncFilter.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
+#include <cstddef>
 #include <cstdint>
 #include <optional>
 
@@ -24,6 +25,15 @@ class ConsensusTransSetSF : public SHAMapSyncFilter
 public:
     using NodeCache = TaggedCache<SHAMapHash, Blob>;
 
+    /**
+     * The size a node's hash-prefixed wire data must reach before gotNode()
+     * tries to parse and resubmit it as a transaction. One byte past the
+     * smallest a hash-prefixed SHAMap leaf can be, which a signed transaction
+     * clears.
+     */
+    static constexpr std::size_t kMinTxNodeBytesToParse =
+        sizeof(std::uint32_t) + kMinShaMapItemBytes + 1;
+
     ConsensusTransSetSF(Application& app, NodeCache& nodeCache);
 
     // Note that the nodeData is overwritten by this call

src/xrpld/app/ledger/InboundLedger.h

@@ -30,9 +30,9 @@
 namespace xrpl {
 
 // A ledger we are trying to acquire
-class InboundLedger final : public TimeoutCounter,
-                            public std::enable_shared_from_this<InboundLedger>,
-                            public CountedObject<InboundLedger>
+class InboundLedger : public TimeoutCounter,
+                      public std::enable_shared_from_this<InboundLedger>,
+                      public CountedObject<InboundLedger>
 {
 public:
     using ClockType = beast::AbstractClock<std::chrono::steady_clock>;
@@ -44,13 +44,29 @@ public:
         CONSENSUS  // We believe the consensus round requires this ledger
     };
 
+    /**
+     * How long to wait between retries, and so how long one timeout takes.
+     */
+    static constexpr std::chrono::milliseconds kRetryInterval{3000};
+
+    /**
+     * @param app The application to run in.
+     * @param hash The ledger to acquire.
+     * @param seq Its sequence, or zero if not known yet.
+     * @param reason Why it is being acquired.
+     * @param clock The clock touch() records against.
+     * @param peerSet Which peers to ask, and how to reach them.
+     * @param retryInterval How long to wait between retries. TimeoutCounter
+     *        requires more than 10ms and less than 30s.
+     */
     InboundLedger(
         Application& app,
         UInt256 const& hash,
         std::uint32_t seq,
         Reason reason,
-        ClockType&,
-        std::unique_ptr<PeerSet> peerSet);
+        ClockType& clock,
+        std::unique_ptr<PeerSet> peerSet,
+        std::chrono::milliseconds retryInterval = kRetryInterval);
 
     ~InboundLedger() override;
 
@@ -59,7 +75,11 @@ public:
     update(std::uint32_t seq);
 
     /**
-     * Returns true if we got all the data.
+     * Whether the acquisition succeeded and its ledger has been settled. Every
+     * path that sets this settles the ledger first, so a caller that sees it
+     * may use the ledger directly.
+     *
+     * @return Whether the ledger is complete and settled.
      */
     bool
     isComplete() const
@@ -68,7 +88,7 @@ public:
     }
 
     /**
-     * Returns false if we failed to get the data.
+     * @return Whether the acquisition has failed.
      */
     bool
     isFailed() const
@@ -76,10 +96,19 @@ public:
         return failed_;
     }
 
+    /**
+     * The acquired ledger.
+     *
+     * A failed acquisition may still hold a partially built ledger, which
+     * getJson() reports on, so ledger_ is kept while this answers nullptr.
+     *
+     * @return The ledger, or nullptr before a header is obtained and once the
+     *         acquisition has failed.
+     */
     std::shared_ptr<Ledger const>
     getLedger() const
     {
-        return ledger_;
+        return failed_ ? nullptr : ledger_;
     }
 
     std::uint32_t
@@ -119,15 +148,35 @@ public:
         return lastAction_;
     }
 
-private:
+protected:
+    // Protected so a test subclass can drive an acquisition through trigger() and done().
+
+    // Why trigger() is being run, which decides how deep a request goes and whether an
+    // aggressive retry applies.
     enum class TriggerReason { Added, Reply, Timeout };
 
+    /**
+     * Ask for more nodes, or judge what has been collected.
+     *
+     * @param peer The peer to ask, or nullptr to ask everyone being tracked.
+     * @param reason Why the acquisition is being triggered.
+     */
+    void
+    trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason);
+
+    /**
+     * Settle the acquisition, publish its outcome, and signal whatever is
+     * waiting on it. Runs at most once. Call under mtx_, which the flags
+     * written here require. Settles before publishing, since isComplete() is
+     * read without mtx_.
+     */
+    void
+    done();
+
+private:
     void
     filterNodes(std::vector<std::pair<SHAMapNodeID, UInt256>>& nodes, TriggerReason reason);
 
-    void
-    trigger(std::shared_ptr<Peer> const&, TriggerReason);
-
     std::vector<NeededHashT>
     getNeededHashes();
 
@@ -137,11 +186,44 @@ private:
     void
     tryDB(node_store::Database& srcDB);
 
-    void
-    done();
+    /**
+     * Whether either map of the ledger being acquired has been found invalid.
+     *
+     * A walk returns a bare list of hashes, so this is what tells a satisfied
+     * map from an abandoned one. See SHAMap::addKnownNode for why the verdict
+     * is final.
+     *
+     * @return Whether either map is Invalid, and false while there is no
+     *         ledger, since then there is no map to judge.
+     */
+    [[nodiscard]] bool
+    hasInvalidMap() const;
+
+    /**
+     * Whether nothing is left to fetch. Not the same as complete: done()
+     * settles the ledger before publishing that.
+     *
+     * @return Whether the header and both maps have been obtained.
+     */
+    [[nodiscard]] bool
+    haveEverything() const
+    {
+        return haveHeader_ && haveState_ && haveTransactions_;
+    }
+
+    /**
+     * Record what one peer's packet achieved, and report its yield. Only a
+     * useful node counts as progress. Call under mtx_.
+     *
+     * @param san What the packet's nodes achieved, accumulated over the whole
+     *        packet.
+     * @return How many good nodes the packet held.
+     */
+    [[nodiscard]] int
+    recordPacket(SHAMapAddNode const& san);
 
     void
-    onTimer(bool progress, ScopedLockType& peerSetLock) override;
+    onTimer(bool progress, ScopedLockType& sl) override;
 
     std::size_t
     getPeerCount() const;
@@ -155,6 +237,17 @@ private:
     bool
     takeHeader(std::string_view data);
 
+    /**
+     * Fail the acquisition when the header's account hash is zero. No ledger
+     * has an empty state map, so such a header cannot name a ledger. Both
+     * tryDB() and takeHeader() judge the header here.
+     *
+     * @return Whether the acquisition was failed. Then failed_ is set and
+     *         ledger_ is null.
+     */
+    bool
+    failOnZeroAccountHash();
+
     void
     receiveNode(
         std::shared_ptr<Peer> const& peer,

src/xrpld/app/ledger/InboundTransactions.h

@@ -39,7 +39,8 @@ public:
      *
      * @param setHash The transaction set ID (digest of the SHAMap root node).
      * @param acquire Whether to fetch the transaction set from the network if
-     * it is missing.
+     *        it is missing. The retention window is refreshed only while the
+     *        acquisition is still worth keeping.
      * @return The transaction set with ID setHash, or nullptr if it is
      * missing.
      */
@@ -50,7 +51,8 @@ public:
      * Add a transaction set from a LedgerData message.
      *
      * @param setHash The transaction set ID (digest of the SHAMap root node).
-     * @param peer The peer that sent the message.
+     * @param peer The peer that sent the message, charged here for a reply
+     *        outside its allowance.
      * @param message The LedgerData message.
      */
     virtual void
@@ -64,11 +66,13 @@ public:
      *
      * @param setHash The transaction set ID (should match set.getHash()).
      * @param set The transaction set.
-     * @param acquired Whether this transaction set was acquired from a peer,
-     * or constructed by ourself during consensus.
+     * @param fromAcquire Whether the acquisition for this hash supplied the set
+     *        itself. False cancels an acquisition still in flight. True leaves
+     *        it registered until newRound() sweeps it, so a late reply still
+     *        reaches the per-peer allowance wantsReplyFrom() judges.
      */
     virtual void
-    giveSet(UInt256 const& setHash, std::shared_ptr<SHAMap> const& set, bool acquired) = 0;
+    giveSet(UInt256 const& setHash, std::shared_ptr<SHAMap> const& set, bool fromAcquire) = 0;
 
     /**
      * Informs the container if a new consensus round

src/xrpld/app/ledger/OpenLedger.h

@@ -240,7 +240,20 @@ OpenLedger::apply(
         auto iter = retries.begin();
         while (iter != retries.end())
         {
-            switch (applyOne(app, view, iter->second, retry, flags, j))
+            // A transaction that cannot be applied counts as a failure, as it does in the pass
+            // above, so it leaves the set and counts as no change. Every entry gets a verdict on
+            // every pass, so the set shrinks.
+            auto result = Result::Failure;
+            try
+            {
+                result = applyOne(app, view, iter->second, retry, flags, j);
+            }
+            catch (std::exception const& e)
+            {
+                JLOG(j.error()) << "OpenLedger::apply: Caught exception: " << e.what();
+            }
+
+            switch (result)
             {
                 case Result::Success:
                     ++changes;

src/xrpld/app/ledger/detail/BuildLedger.cpp

@@ -6,6 +6,7 @@
 
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/chrono.h>
+#include <xrpl/basics/contract.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/ledger/ApplyView.h>
@@ -76,7 +77,18 @@ buildLedgerImpl(
     XRPL_ASSERT(
         built->header().seq < kXrpLedgerEarliestFees || built->read(keylet::feeSettings()),
         "xrpl::buildLedgerImpl : valid ledger fees");
-    built->setAccepted(closeTime, closeResolution, closeTimeCorrect);
+    // The invariant: a ledger this function returns has both maps immutable, which is what
+    // setAccepted() reports on (see Ledger::setImmutable()). Nothing downstream re-checks it.
+    //
+    // logicError() rather than UNREACHABLE(): the consensus caller acts on this ledger straight
+    // away, so the stop has to name this site in every build rather than only where assertions
+    // are on. See RCLConsensus::Adaptor::buildLCL().
+    if (!built->setAccepted(closeTime, closeResolution, closeTimeCorrect))
+    {
+        // LCOV_EXCL_START
+        logicError("buildLedgerImpl: accepted ledger map is invalid");
+        // LCOV_EXCL_STOP
+    }
 
     return built;
 }

src/xrpld/app/ledger/detail/InboundLedger.cpp

@@ -54,8 +54,6 @@
 
 namespace xrpl {
 
-using namespace std::chrono_literals;
-
 static constexpr auto kPeerCountStart = 5;           // Number of peers to start with
 static constexpr auto kPeerCountAdd = 3;             // Number of peers to add on a timeout
 static constexpr auto kLedgerTimeoutRetriesMax = 6;  // how many timeouts before we give up
@@ -65,20 +63,18 @@ static constexpr auto kMissingNodesFind = 256;  // Number of nodes to find initi
 static constexpr auto kReqNodesReply = 128;     // Number of nodes to request for a reply
 static constexpr auto kReqNodes = 12;           // Number of nodes to request blindly
 
-// millisecond for each ledger timeout
-constexpr auto kLedgerAcquireTimeout = 3000ms;
-
 InboundLedger::InboundLedger(
     Application& app,
     UInt256 const& hash,
     std::uint32_t seq,
     Reason reason,
     ClockType& clock,
-    std::unique_ptr<PeerSet> peerSet)
+    std::unique_ptr<PeerSet> peerSet,
+    std::chrono::milliseconds retryInterval)
     : TimeoutCounter(
           app,
           hash,
-          kLedgerAcquireTimeout,
+          retryInterval,
           {.jobType = JtLedgerData, .jobName = "InboundLedger", .jobLimit = 5},
           app.getJournal("InboundLedger"))
     , clock_(clock)
@@ -97,8 +93,12 @@ InboundLedger::init(ScopedLockType& collectionLock)
     collectionLock.unlock();
 
     tryDB(app_.getNodeFamily().db());
+    // done() is what wakes whatever is waiting and records the hash in recentFailures_.
     if (failed_)
+    {
+        done();
         return;
+    }
 
     if (!complete_)
     {
@@ -112,7 +112,21 @@ InboundLedger::init(ScopedLockType& collectionLock)
     XRPL_ASSERT(
         ledger_->header().seq < kXrpLedgerEarliestFees || ledger_->read(keylet::feeSettings()),
         "xrpl::InboundLedger::init : valid ledger fees");
-    ledger_->setImmutable();
+    // tryDB() verified both maps before setting complete_ and mtx_ has been held since, so both
+    // maps are still sound here.
+    if (!ledger_->setImmutable())
+    {
+        // LCOV_EXCL_START
+        // Recorded before the UNREACHABLE, which is not guaranteed to stop here.
+        // Withdrawn as well as failed, to match done() and TransactionAcquire::done(),
+        // for a caller that checks complete_ before failed_.
+        complete_ = false;
+        failed_ = true;
+        done();
+        UNREACHABLE("xrpl::InboundLedger::init : map is invalid");
+        return;
+        // LCOV_EXCL_STOP
+    }
 
     if (reason_ == Reason::HISTORY)
         return;
@@ -222,6 +236,12 @@ InboundLedger::neededStateHashes(int max, SHAMapSyncFilter const* filter) const
     return neededHashes(ledger_->header().accountHash, ledger_->stateMap(), max, filter);
 }
 
+bool
+InboundLedger::hasInvalidMap() const
+{
+    return ledger_ && !ledger_->mapsValid();
+}
+
 // See how much of the ledger data is stored locally
 // Data found in a fetch pack will be stored
 void
@@ -241,7 +261,12 @@ InboundLedger::tryDB(node_store::Database& srcDB)
                     << "hash " << hash_ << " seq " << std::to_string(seq_) << " cannot be a ledger";
                 ledger_.reset();
                 failed_ = true;
+                return;
             }
+
+            // A zero account hash cannot be a ledger either. The helper sets failed_ and drops the
+            // ledger, so the callers below store and publish nothing for such a header.
+            failOnZeroAccountHash();
         };
 
         // Try to fetch the ledger header from the DB
@@ -309,12 +334,6 @@ InboundLedger::tryDB(node_store::Database& srcDB)
 
     if (!haveState_)
     {
-        if (ledger_->header().accountHash.isZero())
-        {
-            JLOG(journal_.fatal()) << "We are acquiring a ledger with a zero account hash";
-            failed_ = true;
-            return;
-        }
         AccountStateSF filter(ledger_->stateMap().family().db(), app_.getLedgerMaster());
         if (ledger_->stateMap().fetchRoot(SHAMapHash{ledger_->header().accountHash}, &filter))
         {
@@ -326,14 +345,34 @@ InboundLedger::tryDB(node_store::Database& srcDB)
         }
     }
 
-    if (haveTransactions_ && haveState_)
+    // Judged here rather than at the setImmutable() below, which runs only once both flags are
+    // set: one map can be abandoned while the other is merely incomplete.
+    if (hasInvalidMap())
+    {
+        JLOG(journal_.warn()) << "Ledger " << hash_ << " found locally has an invalid map";
+        failed_ = true;
+        return;
+    }
+
+    if (haveEverything())
     {
-        JLOG(journal_.debug()) << "Had everything locally";
-        complete_ = true;
         XRPL_ASSERT(
             ledger_->header().seq < kXrpLedgerEarliestFees || ledger_->read(keylet::feeSettings()),
             "xrpl::InboundLedger::tryDB : valid ledger fees");
-        ledger_->setImmutable();
+        // Settled before complete_ is published, so a caller that reads the flag sees a finished
+        // ledger. Reachable despite the guard above, since trigger() walks the state map with
+        // mtx_ released.
+        if (!ledger_->setImmutable())
+        {
+            // LCOV_EXCL_START: only the walk named above reaches this, so no test does
+            JLOG(journal_.warn()) << "Ledger " << hash_ << " found locally is invalid";
+            failed_ = true;
+            return;
+            // LCOV_EXCL_STOP
+        }
+
+        JLOG(journal_.debug()) << "Had everything locally";
+        complete_ = true;
     }
 }
 
@@ -419,6 +458,41 @@ InboundLedger::done()
     signaled_ = true;
     touch();
 
+    // complete_ is published only for a settled, valid ledger, since isComplete() is read without
+    // mtx_. The complete_ arm is for tryDB(), which settles its own result and sets the flag before
+    // calling, so that path arrives with only the reporting below left to do.
+    if (!failed_ && ledger_ && (complete_ || haveEverything()))
+    {
+        XRPL_ASSERT(
+            ledger_->header().seq < kXrpLedgerEarliestFees || ledger_->read(keylet::feeSettings()),
+            "xrpl::InboundLedger::done : valid ledger fees");
+        // trigger() walks the state map with mtx_ released, so the verdict can land after the
+        // flags said there was nothing left to fetch. A race rather than a broken invariant, so
+        // this recovers rather than asserts. setInvalid() outranks Immutable, so an immutable
+        // ledger can still hold an invalid map.
+        SOMETIMES(hasInvalidMap(), "xrpl::InboundLedger::done : map invalidated by a race");
+        if (!ledger_->setImmutable())
+        {
+            JLOG(journal_.warn()) << "Acquired ledger " << hash_ << " is invalid";
+            // Withdrawn as well as failed, for a caller that checks complete_ before failed_.
+            complete_ = false;
+            failed_ = true;
+        }
+        else
+        {
+            complete_ = true;
+            switch (reason_)
+            {
+                case Reason::HISTORY:
+                    app_.getInboundLedgers().onLedgerFetched();
+                    break;
+                default:
+                    app_.getLedgerMaster().storeLedger(ledger_);
+                    break;
+            }
+        }
+    }
+
     JLOG(journal_.debug()) << "Acquire " << hash_ << (failed_ ? " fail " : " ")
                            << ((timeouts_ == 0)
                                    ? std::string()
@@ -427,28 +501,13 @@ InboundLedger::done()
 
     XRPL_ASSERT(complete_ || failed_, "xrpl::InboundLedger::done : complete or failed");
 
-    if (complete_ && !failed_ && ledger_)
-    {
-        XRPL_ASSERT(
-            ledger_->header().seq < kXrpLedgerEarliestFees || ledger_->read(keylet::feeSettings()),
-            "xrpl::InboundLedger::done : valid ledger fees");
-        ledger_->setImmutable();
-        switch (reason_)
-        {
-            case Reason::HISTORY:
-                app_.getInboundLedgers().onLedgerFetched();
-                break;
-            default:
-                app_.getLedgerMaster().storeLedger(ledger_);
-                break;
-        }
-    }
-
-    // We hold the PeerSet lock, so must dispatch
+    // mtx_ is held, so this may only post the work rather than do it.
     app_.getJobQueue().addJob(JtLedgerData, "AcqDone", [self = shared_from_this()]() {
-        if (self->complete_ && !self->failed_)
+        // Read through getLedger(), which consults failed_ itself, so failure and a null ledger_
+        // are both refused by one check. checkAccept() requires a non-null ledger.
+        if (auto const ledger = self->getLedger(); self->complete_ && ledger)
         {
-            self->app_.getLedgerMaster().checkAccept(self->getLedger());
+            self->app_.getLedgerMaster().checkAccept(ledger);
             self->app_.getLedgerMaster().tryAdvance();
         }
         else
@@ -497,6 +556,7 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         if (failed_)
         {
             JLOG(journal_.warn()) << " failed local for " << hash_;
+            done();
             return;
         }
     }
@@ -513,7 +573,20 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         {
             auto need = getNeededHashes();
 
-            if (!need.empty())
+            // The validity check runs ahead of the emptiness test, since getNeededHashes() walks
+            // both maps and can reach the verdict itself. The result is read once, so the hint
+            // below and the test it feeds share one observation of a map another thread can be
+            // invalidating. The claim is withdrawn alongside the failure, so failed_ and
+            // complete_ never both read true.
+            bool const invalidMap = hasInvalidMap();
+            SOMETIMES(invalidMap, "xrpl::InboundLedger::trigger : map is invalid");
+            if (invalidMap)
+            {
+                JLOG(journal_.warn()) << "Acquire " << hash_ << " has an invalid map";
+                failed_ = true;
+                complete_ = false;
+            }
+            else if (!need.empty())
             {
                 protocol::TMGetObjectByHash tmBH;
                 bool typeSet = false;
@@ -550,11 +623,11 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
             }
             else
             {
+                // The tail of this function settles the ledger and reports it complete.
                 JLOG(journal_.info()) << "getNeededHashes says acquire is complete";
                 haveHeader_ = true;
                 haveTransactions_ = true;
                 haveState_ = true;
-                complete_ = true;
             }
         }
     }
@@ -617,27 +690,31 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         {
             AccountStateSF filter(ledger_->stateMap().family().db(), app_.getLedgerMaster());
 
-            // Release the lock while we process the large state map
+            // Release the lock while the state map is walked. mtx_ is recursive, so an sl.unlock()
+            // under onTimer() or the addPeers() callback leaves mtx_ held. The flags are re-read
+            // below because another packet can be handled while this walk runs.
             sl.unlock();
             auto nodes = ledger_->stateMap().getMissingNodes(kMissingNodesFind, &filter);
             sl.lock();
 
+            // The validity check runs outside the flags' guard below, since the verdict is about
+            // the map rather than this round: it holds even if another thread reported the ledger
+            // complete while the lock was released.
+            bool const walkAbandonedMap = hasInvalidMap();
+            SOMETIMES(walkAbandonedMap, "xrpl::InboundLedger::trigger : map abandoned by its walk");
+            if (walkAbandonedMap)
+            {
+                JLOG(journal_.warn()) << "Ledger " << hash_ << " has a map its walk abandoned";
+                failed_ = true;
+                complete_ = false;
+            }
             // Make sure nothing happened while we released the lock
-            if (!failed_ && !complete_ && !haveState_)
+            else if (!failed_ && !complete_ && !haveState_)
             {
                 if (nodes.empty())
                 {
-                    if (!ledger_->stateMap().isValid())
-                    {
-                        failed_ = true;
-                    }
-                    else
-                    {
-                        haveState_ = true;
-
-                        if (haveTransactions_)
-                            complete_ = true;
-                    }
+                    // The test above already caught a map the walk abandoned, so this one is sound.
+                    haveState_ = true;
                 }
                 else
                 {
@@ -699,9 +776,6 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
                 else
                 {
                     haveTransactions_ = true;
-
-                    if (haveState_)
-                        complete_ = true;
                 }
             }
             else
@@ -726,11 +800,12 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         }
     }
 
-    if (complete_ || failed_)
+    // done() settles the ledger before publishing complete_, and requires mtx_, which is still
+    // held here.
+    if (failed_ || haveEverything())
     {
-        JLOG(journal_.debug()) << "Done:" << (complete_ ? " complete" : "")
-                               << (failed_ ? " failed " : " ") << ledger_->header().seq;
-        sl.unlock();
+        JLOG(journal_.debug()) << "Done:" << (failed_ ? " failed " : " have everything ")
+                               << ledger_->header().seq;
         done();
     }
 }
@@ -774,15 +849,31 @@ InboundLedger::filterNodes(
         recentNodes_.insert(n.second);
 }
 
+bool
+InboundLedger::failOnZeroAccountHash()
+{
+    bool const zero = ledger_->header().accountHash.isZero();
+    SOMETIMES(zero, "xrpl::InboundLedger::failOnZeroAccountHash : zero account hash");
+    if (!zero)
+        return false;
+
+    JLOG(journal_.fatal()) << "We are acquiring a ledger with a zero account hash: " << hash_;
+    failed_ = true;
+    ledger_.reset();
+    return true;
+}
+
 /**
- * Take ledger header data
- * Call with a lock
+ * Build the ledger from a header a peer supplied. Call with mtx_ held.
+ *
+ * @param data The serialized header, without a hash prefix.
+ * @return False for a header other than the one asked for. True otherwise,
+ *         also when the header cannot name a ledger, and then failed_ is set,
+ *         ledger_ is null, haveHeader_ stays false, and the caller calls done().
  */
-// data must not have hash prefix
 bool
 InboundLedger::takeHeader(std::string_view data)
 {
-    // Return value: true=normal, false=bad data
     JLOG(journal_.trace()) << "got header acquiring ledger " << hash_;
 
     if (complete_ || failed_ || haveHeader_)
@@ -798,6 +889,12 @@ InboundLedger::takeHeader(std::string_view data)
         ledger_.reset();
         return false;
     }
+
+    // The header is the one asked for, so the peer is not charged and the caller reads failed_.
+    // The helper drops the ledger, and haveHeader_ stays false, as tryDB() leaves them.
+    if (failOnZeroAccountHash())
+        return true;
+
     if (seq_ == 0)
         seq_ = ledger_->header().seq;
     ledger_->stateMap().setLedgerSeq(seq_);
@@ -812,9 +909,6 @@ InboundLedger::takeHeader(std::string_view data)
     if (ledger_->header().txHash.isZero())
         haveTransactions_ = true;
 
-    if (ledger_->header().accountHash.isZero())
-        haveState_ = true;
-
     ledger_->txMap().setSynching();
     ledger_->stateMap().setSynching();
 
@@ -822,8 +916,17 @@ InboundLedger::takeHeader(std::string_view data)
 }
 
 /**
- * Process node data received from a peer
- * Call with a lock
+ * Judge one peer's map-node reply, hooking in what it carries.
+ *
+ * Call with mtx_ held. A node that cannot be hooked costs its sender the
+ * recoverable tier, and one that proves the map impossible costs the harsher
+ * one. The second verdict is read off the batch rather than off the map, since
+ * a getMissingNodes() walk can invalidate the map while this runs.
+ *
+ * @param peer The sender, charged at whichever site judged its data.
+ * @param packet The reply, which names the map it is for.
+ * @param san The running tally, added to as each node is judged, and reset to a
+ *        single rejection if the map turns out to be abandoned.
  */
 void
 InboundLedger::receiveNode(
@@ -903,7 +1006,32 @@ InboundLedger::receiveNode(
             {
                 JLOG(journal_.warn()) << "Got invalid node " << *nodeID << " for ledger " << hash_
                                       << " from peer " << peer->id();
-                peer->charge(resource::kFeeInvalidData, "ledger_node invalid");
+                // The charge test below reads the verdict rather than the map, because a
+                // getMissingNodes() walk runs with mtx_ released (see trigger()) and can
+                // invalidate the map while this packet is judged, and that verdict is not this
+                // sender's doing.
+                if (result.invalidatedMap())
+                {
+                    // This node commits to a shape no valid tree has. See SHAMap::addKnownNode for
+                    // why that verdict holds for every peer.
+                    peer->charge(resource::kFeeMalformedData, "ledger_node makes map invalid");
+                }
+                else
+                {
+                    peer->charge(resource::kFeeInvalidData, "ledger_node invalid");
+                }
+
+                if (!map.isValid())
+                {
+                    // The map is abandoned, whichever walk reached that verdict, so the
+                    // acquisition fails here.
+                    failed_ = true;
+                    done();
+
+                    // The whole packet is discarded, since the nodes ahead of the bad one belong to
+                    // the same impossible tree. Matches TransactionAcquire::takeNodesLocked().
+                    san = SHAMapAddNode::invalid();
+                }
                 return;
             }
         }
@@ -917,6 +1045,19 @@ InboundLedger::receiveNode(
         return;
     }
 
+    // The verdict belongs to whichever walk reached it rather than to this packet, so the
+    // acquisition fails here and the sender keeps its fee. The validity check runs ahead of
+    // isSynching() below, which reads an invalid map the same way as a finished one.
+    if (!map.isValid())
+    {
+        failed_ = true;
+        done();
+
+        // The whole packet is discarded, since its nodes belong to the abandoned map.
+        san = SHAMapAddNode::invalid();
+        return;
+    }
+
     if (!map.isSynching())
     {
         if (packet.type() == protocol::liTX_NODE)
@@ -928,11 +1069,10 @@ InboundLedger::receiveNode(
             haveState_ = true;
         }
 
-        if (haveTransactions_ && haveState_)
-        {
-            complete_ = true;
+        // done() settles the ledger before publishing complete_, so having every part is reported
+        // there rather than here.
+        if (haveEverything())
             done();
-        }
     }
 }
 
@@ -1099,6 +1239,13 @@ InboundLedger::processData(std::shared_ptr<Peer> peer, protocol::TMLedgerData co
                     return -1;
                 }
 
+                // takeHeader() failed the acquisition. Nothing else signals that after isDone().
+                if (failed_)
+                {
+                    done();
+                    return 0;
+                }
+
                 san.incUseful();
             }
 
@@ -1135,11 +1282,7 @@ InboundLedger::processData(std::shared_ptr<Peer> peer, protocol::TMLedgerData co
             return -1;
         }
 
-        if (san.isUseful())
-            progress_ = true;
-
-        stats_ += san;
-        return san.getGood();
+        return recordPacket(san);
     }
 
     if ((packet.type() == protocol::liTX_NODE) || (packet.type() == protocol::liAS_NODE))
@@ -1160,21 +1303,24 @@ InboundLedger::processData(std::shared_ptr<Peer> peer, protocol::TMLedgerData co
                                << ((packet.type() == protocol::liTX_NODE) ? "TX" : "AS")
                                << " node stats: " << san.get();
 
-        // `san` accumulates across the whole packet, so `isInvalid()` (bad_ > 0) does not mean the
-        // packet had no useful nodes: credit whatever good/useful nodes were sent rather than
-        // discarding everything because one node in an otherwise-good packet was bad.
-        // Note: Peer charges for invalid/malformed data are issued from within receiveNode at the
-        // exact failure site, so the peer is only charged for problems they are responsible for.
-        if (san.isUseful())
-            progress_ = true;
-
-        stats_ += san;
-        return san.getGood();
+        // receiveNode() charges the peer at the site that judged the data, and discards the whole
+        // packet only for a node that leaves the map invalid.
+        return recordPacket(san);
     }
 
     return -1;
 }
 
+int
+InboundLedger::recordPacket(SHAMapAddNode const& san)
+{
+    if (san.isUseful())
+        progress_ = true;
+
+    stats_ += san;
+    return san.getGood();
+}
+
 namespace detail {
 // Track the amount of useful data that each peer returns
 struct PeerDataCounts

src/xrpld/app/ledger/detail/InboundTransactions.cpp

@@ -96,11 +96,10 @@ public:
             {
                 if (acquire)
                 {
-                    it->second.seq = seq_;
-                    if (it->second.acquire)
-                    {
-                        it->second.acquire->stillNeed();
-                    }
+                    // Refreshed only while stillNeed() says there is something to wait for, so an
+                    // acquisition whose map is invalid is left for newRound() to sweep.
+                    if (!it->second.acquire || it->second.acquire->stillNeed())
+                        it->second.seq = seq_;
                 }
                 return it->second.set;
             }
@@ -142,6 +141,12 @@ public:
             return;
         }
 
+        // The test runs before the loop below, which costs a hash per node, since a settled
+        // acquisition discards the result. It charges the peer itself when the reply is
+        // outside its allowance.
+        if (!ta->wantsReplyFrom(peer))
+            return;
+
         std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data;
         data.reserve(packet.nodes().size());
 
@@ -168,15 +173,8 @@ public:
             data.emplace_back(*nodeID, std::move(treeNode));
         }
 
-        auto const san = ta->takeNodes(std::move(data), peer);
-        if (san.isInvalid())
-        {
-            peer->charge(resource::kFeeInvalidData, "ledger_data invalid");
-        }
-        else if (!san.isUseful())
-        {
-            peer->charge(resource::kFeeUselessData, "ledger_data useless");
-        }
+        // takeNodes() charges the peer and records the batch itself, so the verdict is discarded.
+        static_cast<void>(ta->takeNodes(std::move(data), peer));
     }
 
     void
@@ -200,7 +198,16 @@ public:
                 inboundSet.set = set;
             }
 
-            inboundSet.acquire.reset();
+            // Reset only when something other than the acquisition supplied the set, since
+            // dropping the pointer cancels an acquisition still in flight. Keeping it lets a late
+            // reply for this hash still reach takeNodesLocked()'s allowance, and newRound() sweeps
+            // the entry.
+            //
+            // Keyed by the hash alone, not by which acquisition finished. addRootNode() refuses a
+            // root that does not hash to the hash asked for, so every acquisition for a hash ends
+            // with the same map, and installing it is correct whichever entry map_[hash] names.
+            if (!fromAcquire)
+                inboundSet.acquire.reset();
         }
 
         if (isNew)

src/xrpld/app/ledger/detail/LedgerPersistence.cpp

@@ -115,8 +115,15 @@ loadLedgerHelper(
     return ledger;
 }
 
+/**
+ * Settle a ledger just loaded from local storage, or discard it.
+ *
+ * @param ledger The ledger to settle. Cleared on failure, so the caller cannot
+ *        hand out one that is still mutable.
+ * @param j Where to log a refusal.
+ */
 static void
-finishLoadByIndexOrHash(std::shared_ptr<Ledger> const& ledger, beast::Journal j)
+finishLoadByIndexOrHash(std::shared_ptr<Ledger>& ledger, beast::Journal j)
 {
     if (!ledger)
         return;
@@ -124,7 +131,18 @@ finishLoadByIndexOrHash(std::shared_ptr<Ledger> const& ledger, beast::Journal j)
     XRPL_ASSERT(
         ledger->header().seq < kXrpLedgerEarliestFees || ledger->read(keylet::feeSettings()),
         "xrpl::finishLoadByIndexOrHash : valid ledger fees");
-    ledger->setImmutable();
+    // Loaded locally. See Ledger::setImmutable().
+    if (!ledger->setImmutable())
+    {
+        // LCOV_EXCL_START
+        JLOG(j.error()) << "Invalid map for ledger " << ledger->header().seq
+                        << "; not marking it as loaded";
+        UNREACHABLE("xrpl::finishLoadByIndexOrHash : map is invalid");
+        // Discarded rather than left un-full, since nothing gates usability on the full flag.
+        ledger.reset();
+        return;
+        // LCOV_EXCL_STOP
+    }
 
     JLOG(j.trace()) << "Loaded ledger: " << to_string(ledger->header().hash);
 

src/xrpld/app/ledger/detail/TimeoutCounter.h

@@ -8,6 +8,7 @@
 
 #include <boost/asio/basic_waitable_timer.hpp>
 
+#include <atomic>
 #include <chrono>
 #include <cstdint>
 #include <memory>
@@ -50,6 +51,9 @@ namespace xrpl {
  * whether to postpone failure and reset the timeout. However, if it can
  * complete all its work in one synchronous step (while it holds the lock), then
  * it can ignore `progress_`.
+ *
+ * `isDone` is not terminal for every subtype: TransactionAcquire::stillNeed()
+ * clears `failed_` and calls `setTimer` again.
  */
 class TimeoutCounter
 {
@@ -98,9 +102,14 @@ protected:
 
     /**
      * Hook called from invokeOnTimer().
+     *
+     * @param progress Whether the subtype recorded progress since the
+     *        last call.
+     * @param sl Proof mtx_ is held. mtx_ is recursive, so a nested lock taken
+     *        inside the call does not release it.
      */
     virtual void
-    onTimer(bool progress, ScopedLockType&) = 0;
+    onTimer(bool progress, ScopedLockType& sl) = 0;
 
     /**
      * Return a weak pointer to this.
@@ -126,10 +135,27 @@ protected:
      */
     UInt256 const hash_;
     int timeouts_{0};
-    bool complete_{false};
-    bool failed_{false};
+
+    // complete_ and failed_ are read without mtx_, so they are atomic rather than guarded: a
+    // reader that sees either flag set also sees the work the writer did before setting it.
+    static_assert(std::atomic<bool>::is_always_lock_free);
+
     /**
-     * Whether forward progress has been made.
+     * Whether the task finished successfully. Each subclass sets it only once
+     * the work it reports on is finished, so a reader may act on it without
+     * taking mtx_.
+     */
+    std::atomic<bool> complete_{false};
+
+    /**
+     * Whether the task gave up.
+     */
+    std::atomic<bool> failed_{false};
+
+    /**
+     * Whether forward progress has been made since invokeOnTimer() last ran.
+     * Each subtype defines what counts as progress, and may read this for
+     * decisions of its own.
      */
     bool progress_{false};
     /**

src/xrpld/app/ledger/detail/TransactionAcquire.cpp

@@ -8,7 +8,9 @@
 
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/base_uint.h>
+#include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/core/Job.h>
+#include <xrpl/resource/Fees.h>
 #include <xrpl/server/NetworkOPs.h>
 #include <xrpl/shamap/SHAMap.h>
 #include <xrpl/shamap/SHAMapAddNode.h>
@@ -18,6 +20,7 @@
 #include <xrpl.pb.h>
 
 #include <algorithm>
+#include <chrono>
 #include <cstddef>
 #include <exception>
 #include <memory>
@@ -26,22 +29,23 @@
 
 namespace xrpl {
 
-using namespace std::chrono_literals;
-
-// Timeout interval in milliseconds
-constexpr auto kTxAcquireTimeout = 250ms;
-
 static constexpr auto kNormTimeouts = 4;
 static constexpr auto kMaxTimeouts = 20;
 
+// How many consecutive timer intervals a duplicate-only reply may postpone. A fan-out race is
+// settled in one or two, so the bound is generous for that case and the timeout count always
+// resumes.
+static constexpr auto kMaxDuplicateCredits = kNormTimeouts;
+
 TransactionAcquire::TransactionAcquire(
     Application& app,
     UInt256 const& hash,
-    std::unique_ptr<PeerSet> peerSet)
+    std::unique_ptr<PeerSet> peerSet,
+    std::chrono::milliseconds retryInterval)
     : TimeoutCounter(
           app,
           hash,
-          kTxAcquireTimeout,
+          retryInterval,
           {.jobType = JtTxnData, .jobName = "TxAcq", .jobLimit = {}},
           app.getJournal("TransactionAcquire"))
     , peerSet_(std::move(peerSet))
@@ -53,16 +57,28 @@ TransactionAcquire::TransactionAcquire(
 void
 TransactionAcquire::done()
 {
-    // We hold a PeerSet lock and so cannot do real work here
-
+    // mtx_ is held, so this may only post real work rather than do it.
     if (failed_)
     {
         JLOG(journal_.debug()) << "Failed to acquire TX set " << hash_;
     }
+    else if (!map_->setImmutable())
+    {
+        // trigger() verified the map before setting complete_ and mtx_ has been held since, and
+        // nothing walks this map with the lock released, so it is still valid. UNREACHABLE for
+        // that reason, with the flags still withdrawn, since UNREACHABLE need not stop here.
+        // LCOV_EXCL_START
+        // Withdraw complete_ alongside the failure, for trigger() and takeNodes(), which both
+        // check complete_ before failed_.
+        complete_ = false;
+        failed_ = true;
+        JLOG(journal_.debug()) << "Failed to acquire TX set " << hash_;
+        UNREACHABLE("xrpl::TransactionAcquire::done : map is invalid");
+        // LCOV_EXCL_STOP
+    }
     else
     {
         JLOG(journal_.debug()) << "Acquired TX set " << hash_;
-        map_->setImmutable();
 
         UInt256 const& hash(hash_);
         std::shared_ptr<SHAMap> const& map(map_);
@@ -79,7 +95,7 @@ TransactionAcquire::done()
 }
 
 void
-TransactionAcquire::onTimer(bool progress, ScopedLockType& psl)
+TransactionAcquire::onTimer(bool progress, ScopedLockType&)
 {
     if (timeouts_ > kMaxTimeouts)
     {
@@ -100,6 +116,31 @@ TransactionAcquire::pmDowncast()
     return shared_from_this();
 }
 
+void
+TransactionAcquire::recordAsked(std::shared_ptr<Peer> const& peer)
+{
+    // Each request renews the pass chargeLateReply() reads, so a peer answering the request
+    // just sent to it is free whatever it answered in an earlier round.
+    if (peer)
+    {
+        requestedPeers_.insert(peer->id());
+        lateReplyGranted_.erase(peer->id());
+        return;
+    }
+
+    // A broadcast goes to every peer the set tracks, so each earns a late-reply pass.
+    auto const& ids = peerSet_->getPeerIds();
+    requestedPeers_.insert(ids.begin(), ids.end());
+    for (auto const id : ids)
+        lateReplyGranted_.erase(id);
+}
+
+bool
+TransactionAcquire::hasAsked(std::shared_ptr<Peer> const& peer) const
+{
+    return requestedPeers_.contains(peer->id());
+}
+
 void
 TransactionAcquire::trigger(std::shared_ptr<Peer> const& peer)
 {
@@ -127,6 +168,7 @@ TransactionAcquire::trigger(std::shared_ptr<Peer> const& peer)
             tmGL.set_querytype(protocol::qtINDIRECT);
 
         *(tmGL.add_nodeids()) = SHAMapNodeID().getRawString();
+        recordAsked(peer);
         peerSet_->sendRequest(tmGL, peer);
     }
     else if (!map_->isValid())
@@ -165,6 +207,7 @@ TransactionAcquire::trigger(std::shared_ptr<Peer> const& peer)
         {
             *tmGL.add_nodeids() = node.first.getRawString();
         }
+        recordAsked(peer);
         peerSet_->sendRequest(tmGL, peer);
     }
 }
@@ -174,24 +217,73 @@ TransactionAcquire::takeNodes(
     std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
     std::shared_ptr<Peer> const& peer)
 {
-    ScopedLockType const sl(mtx_);
+    ScopedLockType sl(mtx_);
 
-    if (complete_)
+    // A settled set accepts no further nodes, and spends this peer's late-reply allowance.
+    if (isDone())
     {
-        JLOG(journal_.trace()) << "TX set complete";
-        return SHAMapAddNode();
+        JLOG(journal_.trace()) << (complete_ ? "TX set complete" : "TX set failed");
+        chargeLateReply(peer, sl);
+
+        // An invalid map's verdict is final, so it outranks the duplicate report.
+        return map_->isValid() ? SHAMapAddNode::duplicate() : SHAMapAddNode::invalid();
     }
 
-    if (failed_)
+    // Read first: the call below can enroll this peer in requestedPeers_.
+    bool const wasAsked = hasAsked(peer);
+
+    auto const san = takeNodesLocked(std::move(data), peer, sl);
+
+    if (san.isUseful())
     {
-        JLOG(journal_.trace()) << "TX set failed";
-        return SHAMapAddNode();
+        // The set advanced, so the duplicate budget is earned back.
+        duplicateCredits_ = 0;
+        progress_ = true;
     }
+    else if (
+        san.getDuplicate() > 0 && san.getBad() == 0 && wasAsked && !progress_ &&
+        duplicateCredits_ < kMaxDuplicateCredits)
+    {
+        // A reply holding duplicates and no bad node, from a peer we asked, counts as an
+        // answer, bounded by kMaxDuplicateCredits and charged at most once per interval.
+        ++duplicateCredits_;
+        progress_ = true;
+    }
+
+    return san;
+}
+
+bool
+TransactionAcquire::wantsReplyFrom(std::shared_ptr<Peer> const& peer)
+{
+    ScopedLockType sl(mtx_);
+
+    if (!isDone())
+        return true;
+
+    JLOG(journal_.trace()) << (complete_ ? "TX set complete" : "TX set failed");
+    chargeLateReply(peer, sl);
+    return false;
+}
+
+SHAMapAddNode
+TransactionAcquire::takeNodesLocked(
+    std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
+    std::shared_ptr<Peer> const& peer,
+    ScopedLockType&)
+{
+    // Accumulated across the batch, so a packet ending in one bad node still counts the nodes
+    // hooked in ahead of it, as InboundLedger::receiveNode() already does.
+    SHAMapAddNode san;
 
     try
     {
         if (data.empty())
+        {
+            // Defensive: PeerImp rejects an empty node list before dispatch.
+            peer->charge(resource::kFeeInvalidData, "tx_set empty");
             return SHAMapAddNode::invalid();
+        }
 
         ConsensusTransSetSF sf(app_, app_.getTempNodeCache());
 
@@ -202,36 +294,70 @@ TransactionAcquire::takeNodes(
                 if (haveRoot_)
                 {
                     JLOG(journal_.debug()) << "Got root TXS node, already have it";
+                    san.incDuplicate();
+                    continue;
                 }
-                else if (!map_->addRootNode(SHAMapHash{hash_}, std::move(d.second), nullptr)
-                              .isGood())
+
+                auto const result =
+                    map_->addRootNode(SHAMapHash{hash_}, std::move(d.second), nullptr);
+                san += result;
+
+                if (!result.isGood())
                 {
                     JLOG(journal_.warn()) << "TX acquire got bad root node for TX set " << hash_
                                           << " from peer " << peer->id();
-                    return SHAMapAddNode::invalid();
-                }
-                else
-                {
-                    haveRoot_ = true;
+                    // addRootNode rejects a hash mismatch and leaves the map valid, so the timer
+                    // owns the retry and picks another peer.
+                    peer->charge(resource::kFeeInvalidData, "tx_set root hash mismatch");
+                    return san;
                 }
+
+                haveRoot_ = true;
+                continue;
             }
-            else if (!map_->addKnownNode(d.first, std::move(d.second), &sf).isGood())
+
+            auto const result = map_->addKnownNode(d.first, std::move(d.second), &sf);
+            san += result;
+
+            if (!result.isGood())
             {
                 JLOG(journal_.warn()) << "TX acquire got bad non-root node " << d.first
                                       << " for TX set " << hash_ << " from peer " << peer->id();
-                return SHAMapAddNode::invalid();
+                if (!map_->isValid())
+                {
+                    // The acquisition fails here, and stillNeed() keeps it failed. See
+                    // SHAMap::addKnownNode for why that verdict holds for every peer. Charged
+                    // kFeeMalformedData under the lock that reached the verdict, so the packet
+                    // that earned the fee is the one that pays it. A deterrent only, since the
+                    // same node reaches a map by paths with no sender to charge.
+                    peer->charge(resource::kFeeMalformedData, "tx_set node makes map invalid");
+                    failed_ = true;
+                    done();
+
+                    // The whole batch is discarded, since the nodes ahead of the bad one belong to
+                    // the same impossible tree, and the acquisition is over.
+                    return SHAMapAddNode::invalid();
+                }
+
+                // Any other bad node leaves the map sound, so the timer owns the retry and picks
+                // another peer.
+                peer->charge(resource::kFeeInvalidData, "tx_set node invalid");
+                return san;
             }
         }
 
         trigger(peer);
-        progress_ = true;
-        return SHAMapAddNode::useful();
+        return san;
     }
     catch (std::exception const& ex)
     {
-        JLOG(journal_.error()) << "Peer " << peer->id()
-                               << " sent us junky transaction node data: " << ex.what();
-        return SHAMapAddNode::invalid();
+        JLOG(journal_.error()) << "TX acquire threw while taking nodes for TX set " << hash_
+                               << " from peer " << peer->id() << ": " << ex.what();
+        // Whatever the batch hooked in before the throw stands, so the tally it reached is what
+        // the caller is told. The timer owns the retry. The sender keeps its fee, since the
+        // classes that reach here are raised by this code rather than by the data.
+        san.incInvalid();
+        return san;
     }
 }
 
@@ -255,12 +381,43 @@ TransactionAcquire::init(int numPeers)
 }
 
 void
+TransactionAcquire::chargeLateReply(std::shared_ptr<Peer> const& peer, ScopedLockType&)
+{
+    if (!hasAsked(peer) || !lateReplyGranted_.insert(peer->id()).second)
+        peer->charge(resource::kFeeUselessData, "tx_set data after the set was settled");
+}
+
+bool
 TransactionAcquire::stillNeed()
 {
-    ScopedLockType const sl(mtx_);
+    ScopedLockType sl(mtx_);
 
     timeouts_ = std::min<int>(timeouts_, kNormTimeouts);
+
+    // A running acquisition keeps the wait it has, rather than restarting it for every consensus
+    // round that asks for the set again.
+    if (!failed_)
+        return true;
+
+    // An invalid map keeps the acquisition failed, since that verdict holds for every peer (see
+    // SHAMap::addKnownNode). Reported so the caller stops refreshing this set's retention window.
+    if (!map_->isValid())
+        return false;
+
     failed_ = false;
+
+    // lateReplyGranted_ is left alone. The free allowance is earned one request at a time, so a
+    // peer keeps the pass it already spent until recordAsked() records another request to it. The
+    // timer restarted below is what sends those requests.
+
+    // The duplicate budget belongs to the round that just failed as well.
+    duplicateCredits_ = 0;
+
+    // Restarting the timer resumes the acquisition. expires_after() cancels whatever wait was
+    // outstanding, so at most one wait is held. A job already handed to the JobQueue still runs one
+    // invokeOnTimer(), which re-arms this same timer and folds back into the one chain.
+    setTimer(sl);
+    return true;
 }
 
 }  // namespace xrpl

src/xrpld/app/ledger/detail/TransactionAcquire.h

@@ -12,8 +12,10 @@
 #include <xrpl/shamap/SHAMapNodeID.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
+#include <chrono>
 #include <cstddef>
 #include <memory>
+#include <set>
 #include <utility>
 #include <vector>
 
@@ -21,41 +23,181 @@ namespace xrpl {
 
 // VFALCO TODO rename to PeerTxRequest
 // A transaction set we are trying to acquire
-class TransactionAcquire final : public TimeoutCounter,
-                                 public std::enable_shared_from_this<TransactionAcquire>,
-                                 public CountedObject<TransactionAcquire>
+class TransactionAcquire : public TimeoutCounter,
+                           public std::enable_shared_from_this<TransactionAcquire>,
+                           public CountedObject<TransactionAcquire>
 {
 public:
     using pointer = std::shared_ptr<TransactionAcquire>;
 
-    TransactionAcquire(Application& app, UInt256 const& hash, std::unique_ptr<PeerSet> peerSet);
+    /**
+     * How long to wait between retries, and so how long one timeout takes.
+     */
+    static constexpr std::chrono::milliseconds kRetryInterval{250};
+
+    /**
+     * @param app The application to run in.
+     * @param hash The set to acquire.
+     * @param peerSet Which peers to ask, and how to reach them.
+     * @param retryInterval How long to wait between retries. TimeoutCounter
+     *        requires more than 10ms and less than 30s.
+     */
+    TransactionAcquire(
+        Application& app,
+        UInt256 const& hash,
+        std::unique_ptr<PeerSet> peerSet,
+        std::chrono::milliseconds retryInterval = kRetryInterval);
     ~TransactionAcquire() override = default;
 
-    SHAMapAddNode
+    /**
+     * Add nodes a peer sent us to the set we are acquiring.
+     *
+     * Takes mtx_, the lock that reaches the verdict, so the charge for
+     * declined data is issued here. A node that leaves the map invalid also
+     * fails the acquisition. A late reply is bounded by the per-peer allowance
+     * in lateReplyGranted_.
+     *
+     * @param data The nodes to add, each with its claimed position.
+     * @param peer The peer that sent them, charged here if the data is
+     *        declined.
+     * @return The tally of useful, duplicate, and bad nodes in the batch.
+     *         Useful and bad can both be nonzero.
+     */
+    [[nodiscard]] SHAMapAddNode
     takeNodes(
         std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
         std::shared_ptr<Peer> const& peer);
 
+    /**
+     * Whether a reply from this peer is worth deserializing.
+     *
+     * Takes mtx_ and spends the peer's late-reply allowance. The answer can go
+     * stale, since mtx_ is released before takeNodes(), which recognizes a late
+     * reply of its own accord.
+     *
+     * @param peer The peer that sent the reply, charged here when the reply is
+     *        outside the allowance.
+     * @return Whether the reply should be parsed and handed to takeNodes().
+     */
+    [[nodiscard]] bool
+    wantsReplyFrom(std::shared_ptr<Peer> const& peer);
+
     void
     init(int startPeers);
 
-    void
+    /**
+     * Resume a timed-out acquisition, or leave it alone.
+     *
+     * Takes mtx_. Clamps the timeout count, then clears the failed flag and
+     * restarts the timer if the map is still valid. An invalid map stays
+     * failed. See SHAMap::addKnownNode for why that verdict is final.
+     *
+     * @return Whether the set is still worth keeping, which is false once its
+     *         map is invalid.
+     */
+    [[nodiscard]] bool
     stillNeed();
 
-private:
+protected:
+    // Protected so a test subclass can read the map's state.
     std::shared_ptr<SHAMap> map_;
+
+private:
     bool haveRoot_{false};
+
+    /**
+     * Every peer a request has actually been sent to, which is what earns a
+     * peer the free late-reply pass. trigger() records a peer where it builds
+     * the request, so selection alone does not enroll one. Survives a
+     * stillNeed() revival. Guarded by mtx_.
+     */
+    std::set<Peer::ID> requestedPeers_;
+
+    /**
+     * Peers in requestedPeers_ that have spent the free late reply their last
+     * request earned. One unspent pass per peer, renewed by each request sent
+     * to it, so overlapping requests to one peer share one pass. recordAsked()
+     * renews a pass wherever it records a request, for a targeted request and
+     * for a broadcast alike, so a revival on its own renews nothing. Guarded
+     * by mtx_.
+     */
+    std::set<Peer::ID> lateReplyGranted_;
+
+    /**
+     * Consecutive timer intervals a duplicate-only reply has postponed, with no
+     * useful node in between. Bounded by kMaxDuplicateCredits, so the timeout
+     * count always resumes. Reset by a batch that advances the set, and by
+     * stillNeed() on revival.
+     */
+    int duplicateCredits_{0};
+
     std::unique_ptr<PeerSet> peerSet_;
 
-    void
-    onTimer(bool progress, ScopedLockType& peerSetLock) override;
+    /**
+     * Add nodes a peer sent us, on the lock takeNodes() holds. Reached only for
+     * an acquisition still running.
+     *
+     * @param data The nodes to add, each with its claimed position.
+     * @param peer The peer that sent them, charged here if the data is
+     *        declined.
+     * @param sl Proof mtx_ is held.
+     * @return The tally of useful, duplicate, and bad nodes in the batch.
+     */
+    SHAMapAddNode
+    takeNodesLocked(
+        std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
+        std::shared_ptr<Peer> const& peer,
+        ScopedLockType& sl);
 
+    /**
+     * Spend this peer's one free late reply, or charge it for replaying. Only
+     * one of wantsReplyFrom() and takeNodes() sees any given reply, so a reply
+     * is charged once.
+     *
+     * @param peer The peer that sent the reply.
+     * @param sl Proof mtx_ is held, which the allowance sets require.
+     */
+    void
+    chargeLateReply(std::shared_ptr<Peer> const& peer, ScopedLockType& sl);
+
+    void
+    onTimer(bool progress, ScopedLockType& sl) override;
+
+    /**
+     * Settle the acquired set and hand it on, or report the failure. Call under
+     * mtx_. Runs once per outcome, and a stillNeed() revival gives one object a
+     * second outcome.
+     */
     void
     done();
 
     void
     addPeers(std::size_t limit);
 
+    /**
+     * Record every peer a request has just gone to, and renew each one's free
+     * late reply. Call under mtx_.
+     *
+     * The renewal gives each peer one unspent pass, renewed by each request
+     * sent to it, so overlapping requests to one peer share one pass. See
+     * lateReplyGranted_.
+     *
+     * @param peer The peer a targeted request went to, or nullptr for a
+     *        broadcast, which reaches every peer the set tracks.
+     */
+    void
+    recordAsked(std::shared_ptr<Peer> const& peer);
+
+    /**
+     * Whether a request has ever gone to this peer, in any round. Call under
+     * mtx_.
+     *
+     * @param peer The peer to ask about, which must be non-null.
+     * @return Whether a request was sent to this peer.
+     */
+    [[nodiscard]] bool
+    hasAsked(std::shared_ptr<Peer> const& peer) const;
+
     void
     trigger(std::shared_ptr<Peer> const&);
     std::weak_ptr<TimeoutCounter>

src/xrpld/app/main/Application.cpp

@@ -1702,7 +1702,14 @@ ApplicationImp::startGenesisLedger()
     XRPL_ASSERT(
         next->header().seq < kXrpLedgerEarliestFees || next->read(keylet::feeSettings()),
         "xrpl::ApplicationImp::startGenesisLedger : valid ledger fees");
-    next->setImmutable();
+    // Built locally. See Ledger::setImmutable(). Failed here rather than at storeLedger() below,
+    // which would name the wrong site.
+    if (!next->setImmutable())
+    {
+        // LCOV_EXCL_START
+        logicError("startGenesisLedger: genesis ledger map is invalid");
+        // LCOV_EXCL_STOP
+    }
     openLedger_.emplace(next, cachedSLEs_, logs_->journal("OpenLedger"));
     ledgerMaster_->storeLedger(next);
     ledgerMaster_->switchLCL(next);
@@ -1724,7 +1731,16 @@ ApplicationImp::getLastFullLedger()
         XRPL_ASSERT(
             ledger->header().seq < kXrpLedgerEarliestFees || ledger->read(keylet::feeSettings()),
             "xrpl::ApplicationImp::getLastFullLedger : valid ledger fees");
-        ledger->setImmutable();
+        // Loaded locally. See Ledger::setImmutable().
+        if (!ledger->setImmutable())
+        {
+            // LCOV_EXCL_START
+            JLOG(j.error()) << "Last full ledger " << seq << " has an invalid map; ignoring it";
+            UNREACHABLE("xrpl::ApplicationImp::getLastFullLedger : map is invalid");
+            // Must not fall through: that would mark a damaged ledger validated.
+            return {};
+            // LCOV_EXCL_STOP
+        }
 
         if (getLedgerMaster().haveLedger(seq))
             ledger->setValidated();
@@ -1876,7 +1892,16 @@ ApplicationImp::loadLedgerFromFile(std::string const& name)
             loadLedger->header().seq < kXrpLedgerEarliestFees ||
                 loadLedger->read(keylet::feeSettings()),
             "xrpl::ApplicationImp::loadLedgerFromFile : valid ledger fees");
-        loadLedger->setAccepted(closeTime, closeTimeResolution, !closeTimeEstimated);
+        // Built locally. See Ledger::setImmutable(). The caller handles a failure return, so this
+        // returns rather than treating it as unreachable.
+        if (!loadLedger->setAccepted(closeTime, closeTimeResolution, !closeTimeEstimated))
+        {
+            // LCOV_EXCL_START
+            JLOG(journal_.fatal()) << "Ledger from file has an invalid map";
+            UNREACHABLE("xrpl::ApplicationImp::loadLedgerFromFile : map is invalid");
+            return nullptr;
+            // LCOV_EXCL_STOP
+        }
 
         return loadLedger;
     }