fix: Make the SHAMap sync-path state atomic

Background ledger acquisition reads and writes SHAMap::state_, ::full_,
::ledgerSeq_, and SHAMapInnerNode::fullBelowGen_ concurrently with the
thread driving it, so all four are now std::atomic. finishFetch() withdraws
full_ with an exchange behind a relaxed load, so exactly one reader thread
reports a gap; ledgerSeq_ stays relaxed both ways since it's only a
nodestore lookup hint.

Ledger::setFull() sets each map's sequence before its full flag, so the
release/exchange ordering makes the sequence visible to whichever thread's
exchange wins the gap report.

.cspell.config.yaml

@@ -102,6 +102,7 @@ words:
   - dearmor
   - decryptor
   - dedented
+  - dedup
   - deleteme
   - demultiplexer
   - deserializaton
@@ -341,6 +342,7 @@ words:
   - unambiguity
   - unauthorizes
   - unauthorizing
+  - undeserializable
   - unergonomic
   - unfetched
   - unfindable

.github/workflows/build-packaging-images.yml

@@ -33,14 +33,12 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        # Newest of each distro: these images only wrap pre-built binaries, so
-        # they set no floor for consumers. build_pkg.py pins the RPM dist tag.
         distro:
           - name: debian
-            base_image: debian:trixie
-          # AlmaLinux rather than UBI, which does not ship rpm-sign.
+            base_image: debian:bookworm
+          # AlmaLinux rather than UBI9, which does not ship rpm-sign.
           - name: rhel
-            base_image: almalinux:10
+            base_image: almalinux:9
     uses: XRPLF/actions/.github/workflows/build-multiarch-image.yml@65d5a0bd72be4ecea95cff0673a6e0672ab5243a
     with:
       image_name: xrpld/packaging-${{ matrix.distro.name }}

include/xrpl/beast/net/IPAddress.h

@@ -103,7 +103,7 @@ namespace boost {
 template <>
 struct hash<::beast::ip::Address>
 {
-    hash() = default;
+    explicit hash() = default;
 
     std::size_t
     operator()(::beast::ip::Address const& addr) const

include/xrpl/ledger/Ledger.h

@@ -285,10 +285,13 @@ public:
     void
     setFull() const
     {
-        txMap_.setFull();
+        // Sequence before flag, per map: setLedgerSeq() stores relaxed and setFull() stores
+        // release, so only this order publishes the sequence to SHAMap::finishFetch(), which
+        // reads it after winning the exchange on the flag.
         txMap_.setLedgerSeq(header_.seq);
-        stateMap_.setFull();
+        txMap_.setFull();
         stateMap_.setLedgerSeq(header_.seq);
+        stateMap_.setFull();
     }
 
     void

include/xrpl/protocol/Book.h

@@ -133,7 +133,7 @@ private:
     using id_hash_type = boost::base_from_member<std::hash<xrpl::MPTID>, 0>;
 
 public:
-    hash() = default;
+    explicit hash() = default;
 
     using value_type = std::size_t;
     using argument_type = xrpl::MPTIssue;
@@ -160,7 +160,7 @@ private:
     mptissue_hasher mMptissueHasher_;
 
 public:
-    hash() = default;
+    explicit hash() = default;
 
     value_type
     operator()(argument_type const& asset) const
@@ -227,7 +227,7 @@ struct hash<xrpl::Issue> : std::hash<xrpl::Issue>
 template <>
 struct hash<xrpl::MPTIssue> : std::hash<xrpl::MPTIssue>
 {
-    hash() = default;
+    explicit hash() = default;
 
     using Base = std::hash<xrpl::MPTIssue>;
 };
@@ -235,7 +235,7 @@ struct hash<xrpl::MPTIssue> : std::hash<xrpl::MPTIssue>
 template <>
 struct hash<xrpl::Asset> : std::hash<xrpl::Asset>
 {
-    hash() = default;
+    explicit hash() = default;
 
     using Base = std::hash<xrpl::Asset>;
 };

include/xrpl/protocol/MPTIssue.h

@@ -151,7 +151,7 @@ namespace std {
 template <>
 struct hash<xrpl::MPTID> : xrpl::MPTID::hasher
 {
-    hash() = default;
+    explicit hash() = default;
 };
 
 }  // namespace std

include/xrpl/shamap/SHAMap.h

@@ -16,6 +16,7 @@
 #include <xrpl/shamap/SHAMapMissingNode.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
+#include <atomic>
 #include <condition_variable>
 #include <cstddef>
 #include <cstdint>
@@ -40,7 +41,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.
      *
@@ -120,16 +121,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 when a map's ledger sequence is established (Ledger::setFull(),
+     * InboundLedger) while a nodestore reader thread reads it. Relaxed either
+     * way: it only serves as a lookup hint for a nodestore keyed by hash, so
+     * it orders nothing else.
      */
-    std::uint32_t ledgerSeq_ = 0;
+    std::atomic<std::uint32_t> ledgerSeq_ = 0;
 
     SHAMapTreeNodePtr root_;
-    mutable SHAMapState state_;
+
+    /**
+     * The map's state.
+     *
+     * A getMissingNodes() walk writes it, through clearSynching(), while whatever
+     * drives the acquisition reads it. Nothing here requires the caller to hold a
+     * lock across the walk, and the acquisition code does not, so this is atomic
+     * rather than guarded.
+     */
+    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.
+     *
+     * finishFetch() clears it on whichever nodestore reader thread completes a
+     * read - several at once for the reads a getMissingNodes() walk posts.
+     */
+    mutable std::atomic<bool> full_ = false;
 
 public:
     /**
@@ -375,16 +403,33 @@ public:
         SHAMapTreeNodePtr treeNode,
         SHAMapSyncFilter const* filter);
 
-    // status functions
     void
     setImmutable();
-    bool
+
+    /**
+     * Whether the map's hash is fixed while nodes may still be added to it.
+     *
+     * @return Whether the map is being synced against a hash it was given.
+     */
+    [[nodiscard]] bool
     isSynching() const;
     void
     setSynching();
     void
     clearSynching();
-    bool
+
+    /**
+     * Whether the map can still be the map it claims to be.
+     *
+     * Not "complete" and not "self-consistent": a map that is merely missing
+     * nodes is valid, and stays valid until something proves the tree it is
+     * syncing against cannot exist. Only the map itself reaches that
+     * verdict, and only from a node the hashes it was given cannot
+     * accommodate.
+     *
+     * @return Whether the map has not been proven impossible.
+     */
+    [[nodiscard]] bool
     isValid() const;
 
     // caution: otherMap must be accessed only by this function
@@ -424,6 +469,36 @@ private:
     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 are still mutated without
+     * ordering guarantees, so this says nothing about whether the rest of
+     * the map is safe to read concurrently.
+     *
+     * @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.
+     *
+     * Private because only the map itself can prove that, from a node that
+     * contradicts the hashes it is syncing against.
+     */
+    void
+    setInvalid();
+
     // tree node cache operations
     SHAMapTreeNodePtr
     cacheLookup(SHAMapHash const& hash) const;
@@ -639,44 +714,62 @@ 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 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 void
 SHAMap::setImmutable()
 {
-    XRPL_ASSERT(state_ != SHAMapState::Invalid, "xrpl::SHAMap::setImmutable : state is valid");
-    state_ = SHAMapState::Immutable;
+    XRPL_ASSERT(isValid(), "xrpl::SHAMap::setImmutable : state is valid");
+    state_.store(SHAMapState::Immutable, std::memory_order_release);
 }
 
 inline bool
 SHAMap::isSynching() const
 {
-    return state_ == SHAMapState::Synching;
+    return state() == SHAMapState::Synching;
 }
 
 inline void
 SHAMap::setSynching()
 {
-    state_ = SHAMapState::Synching;
+    state_.store(SHAMapState::Synching, std::memory_order_release);
 }
 
 inline void
 SHAMap::clearSynching()
 {
-    state_ = SHAMapState::Modifying;
+    state_.store(SHAMapState::Modifying, std::memory_order_release);
 }
 
 inline bool
 SHAMap::isValid() const
 {
-    return state_ != SHAMapState::Invalid;
+    return state() != SHAMapState::Invalid;
+}
+
+inline void
+SHAMap::setInvalid()
+{
+    state_.store(SHAMapState::Invalid, std::memory_order_release);
 }
 
 inline void

include/xrpl/shamap/SHAMapAddNode.h

@@ -14,35 +14,140 @@ private:
 
 public:
     SHAMapAddNode();
+
+    /**
+     * Record one node that was rejected.
+     *
+     * Counted rather than merely flagged, so a batch that carries on past a
+     * rejected node reports one per node instead of just that it happened.
+     */
     void
     incInvalid();
+
+    /**
+     * Record one node that was hooked into the map.
+     *
+     * Counted so a batch's tally can be read back through getGood().
+     */
     void
     incUseful();
+
+    /**
+     * Record one node the map already held.
+     *
+     * Counted separately from a useful node: it is not new data, but it is
+     * also not a rejection - see isGood().
+     */
     void
     incDuplicate();
-    void
-    reset();
+
+    /**
+     * How many nodes were hooked into the map, which isUseful() only reports
+     * the presence of.
+     *
+     * @return The count.
+     */
     [[nodiscard]] int
     getGood() const;
+    /**
+     * How many nodes were rejected, which isInvalid() only reports the presence
+     * of. A batch that stops on its first bad node counts one; one that carries
+     * on counts each.
+     *
+     * @return The count.
+     */
+    [[nodiscard]] int
+    getBad() const;
+
+    /**
+     * How many nodes the batch already held, which no other accessor reports: a
+     * duplicate counts as neither good nor bad.
+     *
+     * @return The count.
+     */
+    [[nodiscard]] int
+    getDuplicate() const;
+
+    /**
+     * Whether the batch overall was worth the exchange: nodes accepted or
+     * already held outnumber the ones rejected.
+     *
+     * A duplicate counts on the accepted side, since the peer answered a
+     * request rather than sent something unasked for; see incDuplicate().
+     *
+     * @return Whether the batch was good.
+     */
     [[nodiscard]] bool
     isGood() const;
+
+    /**
+     * Whether any node in the batch was rejected.
+     *
+     * @return Whether at least one node was bad.
+     */
     [[nodiscard]] bool
     isInvalid() const;
+
+    /**
+     * Whether any node in the batch was hooked into the map.
+     *
+     * @return Whether at least one node was useful.
+     */
     [[nodiscard]] bool
     isUseful() const;
+
+    /**
+     * A verdict recording one duplicate node.
+     *
+     * @return The verdict.
+     */
+    static SHAMapAddNode
+    duplicate();
+
+    /**
+     * A verdict recording one useful node.
+     *
+     * @return The verdict.
+     */
+    static SHAMapAddNode
+    useful();
+
+    /**
+     * A verdict recording one invalid node.
+     *
+     * @return The verdict.
+     */
+    static SHAMapAddNode
+    invalid();
+
+    /**
+     * Clear every count back to zero.
+     */
+    void
+    reset();
+
+    /**
+     * Render the tally as a log line.
+     *
+     * A format rather than an API: a caller that needs the counts themselves
+     * should read them through getGood(), getBad() and getDuplicate() instead
+     * of parsing this.
+     *
+     * @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);
 
-    static SHAMapAddNode
-    duplicate();
-    static SHAMapAddNode
-    useful();
-    static SHAMapAddNode
-    invalid();
-
 private:
     SHAMapAddNode(int good, int bad, int duplicate);
 };
@@ -74,18 +179,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,22 +215,6 @@ 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
-{
-    return (good_ + duplicate_) > bad_;
-}
-
 inline SHAMapAddNode
 SHAMapAddNode::duplicate()
 {
@@ -132,6 +233,12 @@ SHAMapAddNode::invalid()
     return SHAMapAddNode(0, 1, 0);
 }
 
+inline void
+SHAMapAddNode::reset()
+{
+    good_ = bad_ = duplicate_ = 0;
+}
+
 inline std::string
 SHAMapAddNode::get() const
 {
@@ -160,4 +267,14 @@ SHAMapAddNode::get() const
     return ret;
 }
 
+inline SHAMapAddNode&
+SHAMapAddNode::operator+=(SHAMapAddNode const& n)
+{
+    good_ += n.good_;
+    bad_ += n.bad_;
+    duplicate_ += n.duplicate_;
+
+    return *this;
+}
+
 }  // namespace xrpl

include/xrpl/shamap/SHAMapInnerNode.h

@@ -31,7 +31,25 @@ private:
      */
     TaggedPointer hashesAndChildren_;
 
-    std::uint32_t fullBelowGen_ = 0;
+    // Inner nodes are allocated in the millions into a deliberately packed layout, so pin that
+    // wrapping fullBelowGen_ leaves every member at the offset it had, and that isFullBelow() does
+    // not take a mutex once per node of every walk.
+    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));
+
+    /**
+     * Written from more than one thread: canonicalization shares nodes between
+     * maps, so concurrent walks of different maps reach the same node, and a
+     * single map's walk can run with the acquisition lock released (see
+     * SHAMap::state_).
+     *
+     * Relaxed both ways: a generation is only ever compared for equality
+     * and publishes nothing, since the children it vouches for are
+     * published through lock_. Ordering it would cost a barrier per inner
+     * node of every walk.
+     */
+    std::atomic<std::uint32_t> fullBelowGen_ = 0;
     std::uint16_t isBranch_ = 0;
 
     /**
@@ -204,13 +222,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

package/install-packaging-tools.sh

@@ -28,31 +28,31 @@ esac
 #   - debhelper and dpkg-dev build the DEB
 #   - rpm-build builds the RPM, with systemd-rpm-macros and redhat-rpm-config
 #     supplying the systemd and find-debuginfo macros the spec uses
-#   - rpm-sign and gnupg2 sign the built RPM
-#   - python3 runs the packaging scripts
-#   - git gives build_pkg.py the commit timestamp it stamps files with
-#   - ca-certificates lets git and the packaging scripts verify TLS
+#   - rpm-sign signs the built RPM
+#   - git gives build_pkg.sh a real history to read SOURCE_DATE_EPOCH from;
+#     without one the timestamp falls back to the wall clock
+#   - curl uploads the finished packages in publish_pkg.sh
+#   - ca-certificates lets curl and git verify TLS
 function install() {
     case "${ID}" in
         debian | ubuntu)
             apt-get update -y
             apt-get install -y --no-install-recommends \
                 ca-certificates \
+                curl \
                 debhelper \
                 debhelper-compat \
                 dpkg-dev \
-                git \
-                python3
+                git
             ;;
 
         rhel | centos | rocky | almalinux)
             dnf install -y --setopt=install_weak_deps=False \
+                curl-minimal \
                 git \
-                gnupg2 \
-                python3 \
-                redhat-rpm-config \
                 rpm-build \
                 rpm-sign \
+                redhat-rpm-config \
                 systemd-rpm-macros
             ;;
     esac

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,14 @@ 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_)
 {
     // If either map may change, they cannot share nodes
-    if ((state_ != SHAMapState::Immutable) || (other.state_ != SHAMapState::Immutable))
+    if ((state() != SHAMapState::Immutable) || (other.state() != SHAMapState::Immutable))
     {
         unshare();
     }
@@ -105,7 +106,7 @@ SHAMap::dirtyUp(SharedPtrNodeStack& stack, uint256 const& target, SHAMapTreeNode
     // 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");
 
@@ -165,7 +166,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);
 }
 
@@ -178,10 +179,15 @@ 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 exchange
+            // rather than a test-then-clear pair is what leaves only one of the reader threads
+            // that miss reporting; the relaxed load ahead of it keeps a map that is already not
+            // full off the exclusive-write path, since full_ shares a cache line with state_ and
+            // ledgerSeq_ and a walk posts up to 512 reads per pass.
+            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 {};
         }
@@ -214,7 +220,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);
             }
@@ -394,7 +400,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);
@@ -419,7 +425,7 @@ SHAMap::unshareNode(intr_ptr::SharedPtr<Node> node, SHAMapNodeID const& nodeID)
     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())
             root_ = node;
@@ -673,7 +679,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");
 
     SharedPtrNodeStack stack;
     walkTowardsKey(id, &stack);
@@ -755,7 +761,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
@@ -846,7 +852,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");
 
     SharedPtrNodeStack stack;
     walkTowardsKey(tag, &stack);
@@ -937,7 +943,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,10 @@ 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: the generation is only ever compared for equality, and 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/SHAMapSync.cpp

@@ -555,7 +555,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();
@@ -623,7 +623,7 @@ SHAMap::addKnownNode(
             (treeNode->isInner() && currNodeID.getDepth() == kLeafDepth))
         {
             // Map is provably invalid
-            state_ = SHAMapState::Invalid;
+            setInvalid();
             return SHAMapAddNode::useful();
         }
 
@@ -647,7 +647,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();

src/test/app/AcquireTestHelpers.h

@@ -0,0 +1,464 @@
+#pragma once
+
+#include <xrpld/app/ledger/ConsensusTransSetSF.h>
+#include <xrpld/overlay/Message.h>
+#include <xrpld/overlay/Peer.h>
+#include <xrpld/overlay/PeerSet.h>
+
+#include <xrpl/basics/SHAMapHash.h>
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/beast/net/IPEndpoint.h>
+#include <xrpl/json/json_value.h>
+#include <xrpl/protocol/KeyType.h>
+#include <xrpl/protocol/PublicKey.h>
+#include <xrpl/protocol/SecretKey.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
+// why the protobuf reply builder below cannot be.
+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 is never taken for a transaction and
+// resubmitted.
+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, and is otherwise inert.
+ *
+ * One instance per packet keeps charges() unambiguous about which packet was
+ * charged what.
+ *
+ * charges_ is unguarded: nothing on the retry timer's path charges a peer, so
+ * every charge lands on the thread that fed the packet in.
+ */
+class ChargeRecordingPeer : public Peer
+{
+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) : id_(nextId()), hasTxSet_(hasTxSet)
+    {
+    }
+
+    void
+    charge(resource::Charge const& fee, std::string const& context = {}) override
+    {
+        charges_.push_back(fee);
+    }
+
+    [[nodiscard]] std::vector<resource::Charge> const&
+    charges() const
+    {
+        return charges_;
+    }
+
+    [[nodiscard]] id_t
+    id() const override
+    {
+        return id_;
+    }
+
+    [[nodiscard]] bool
+    hasTxSet(uint256 const&) const override
+    {
+        return hasTxSet_;
+    }
+
+    // Nothing below is consulted by the paths under test.
+
+    void
+    send(std::shared_ptr<Message> const&) override
+    {
+    }
+    [[nodiscard]] beast::ip::Endpoint
+    getRemoteAddress() const override
+    {
+        return {};
+    }
+    [[nodiscard]] bool
+    cluster() const override
+    {
+        return false;
+    }
+    [[nodiscard]] bool
+    isHighLatency() const override
+    {
+        return false;
+    }
+    [[nodiscard]] int
+    getScore(bool) const override
+    {
+        return 0;
+    }
+    [[nodiscard]] PublicKey const&
+    getNodePublic() const override
+    {
+        // Shared across instances: nothing tells these peers apart by key, and deriving one
+        // per instance runs a real Ed25519 keygen for every peer a case builds.
+        static PublicKey const kNodePublicKey =
+            derivePublicKey(KeyType::Ed25519, randomSecretKey());
+        return kNodePublicKey;
+    }
+    json::Value
+    json() override
+    {
+        return {};
+    }
+    [[nodiscard]] bool
+    supportsFeature(ProtocolFeature) const override
+    {
+        return false;
+    }
+    [[nodiscard]] std::optional<std::size_t>
+    publisherListSequence(PublicKey const&) const override
+    {
+        return {};
+    }
+    void
+    setPublisherListSequence(PublicKey const&, std::size_t const) override
+    {
+    }
+    [[nodiscard]] uint256
+    getClosedLedgerHash() const override
+    {
+        static uint256 const kHash{};
+        return kHash;
+    }
+    [[nodiscard]] bool
+    hasLedger(uint256 const&, std::uint32_t) const override
+    {
+        return true;
+    }
+    void
+    ledgerRange(std::uint32_t&, std::uint32_t&) const override
+    {
+    }
+    void
+    cycleStatus() override
+    {
+    }
+    bool
+    hasRange(std::uint32_t, std::uint32_t) override
+    {
+        return false;
+    }
+    [[nodiscard]] bool
+    compressionEnabled() const override
+    {
+        return false;
+    }
+    void
+    sendTxQueue() override
+    {
+    }
+    void
+    addTxQueue(uint256 const&) override
+    {
+    }
+    void
+    removeTxQueue(uint256 const&) override
+    {
+    }
+    [[nodiscard]] bool
+    txReduceRelayEnabled() const override
+    {
+        return false;
+    }
+    [[nodiscard]] std::string const&
+    fingerprint() const override
+    {
+        static std::string const kFingerprint;
+        return kFingerprint;
+    }
+
+private:
+    /**
+     * The next id to hand out, distinct per instance so a test with
+     * several peers can tell from a recorded id which one an acquisition
+     * picked.
+     *
+     * @return The id.
+     */
+    [[nodiscard]] static id_t
+    nextId()
+    {
+        static std::atomic<id_t> next{1};
+        return next++;
+    }
+
+    std::vector<resource::Charge> charges_;
+    id_t id_;
+    bool hasTxSet_;
+};
+
+/**
+ * A peer set that counts the requests sent through it, which is what shows
+ * whether an acquisition is still asking for nodes, and that offers peers to a
+ * hasItem/onPeerAdded callback pair like the real one: hard-filtered by hasItem
+ * (which only scores in the real peer set) and deduped by tracked id, same as
+ * the real peer set.
+ *
+ * The retry timer drives addPeers() and sendRequest() from a job thread while
+ * the test reads the results, so everything recorded here is guarded.
+ */
+class RequestCountingPeerSet : public PeerSet
+{
+public:
+    /**
+     * @param candidates The peers addPeers() may offer, in the order they are
+     *        considered. Fixed at construction, so nothing can change them
+     *        while an acquisition is running. Empty for a case that never lets
+     *        addPeers() find anyone.
+     */
+    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 so a test can check what was
+     *        asked for.
+     * @param hasItem Hard-filters the candidates worth asking. Selects rather than merely
+     *        scores, unlike the real peer set's use of the same parameter.
+     * @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: a candidate already selected by an
+                // earlier call does not get offered - or its onPeerAdded rerun - again.
+                if (hasItem(candidate) && addedPeers_.insert(candidate->id()).second)
+                    selected.push_back(candidate);
+            }
+        }
+
+        // Outside the lock: onPeerAdded() calls back into the acquisition, which sends a
+        // request straight back through this object.
+        for (auto const& peer : selected)
+            onPeerAdded(peer);
+    }
+
+    void
+    sendRequest(
+        ::google::protobuf::Message const&,
+        protocol::MessageType,
+        std::shared_ptr<Peer> const& peer) override
+    {
+        std::scoped_lock const lock(mutex_);
+        ++requests_;
+    }
+
+    /**
+     * The ids of every peer addPeers() has selected, which is what an
+     * acquisition takes for the peers it is tracking.
+     *
+     * Unguarded, like the real peer set's: every caller of this and of
+     * addPeers() is an acquisition holding its own mtx_, so nothing can be
+     * added while a caller iterates. The by-value accessor below is what the
+     * test thread reads instead.
+     *
+     * A caveat for a case that wants a peer count rather than a set of ids:
+     * InboundLedger::getPeerCount() resolves each id through
+     * Overlay::findPeerByShortID(), which only knows peers that really
+     * connected, so it still reports zero however many ids are returned here.
+     *
+     * @return The ids.
+     */
+    [[nodiscard]] std::set<Peer::id_t> const&
+    getPeerIds() const override
+    {
+        return addedPeers_;
+    }
+
+    [[nodiscard]] int
+    requests() const
+    {
+        std::scoped_lock const lock(mutex_);
+        return requests_;
+    }
+
+    /**
+     * The limit the first addPeers() call asked for.
+     *
+     * The first rather than the last, because onTimer() keeps calling
+     * addPeers(1) for as long as an acquisition runs, which would overwrite
+     * what init() asked for.
+     */
+    [[nodiscard]] std::optional<std::size_t>
+    firstLimit() const
+    {
+        std::scoped_lock const lock(mutex_);
+        return firstLimit_;
+    }
+
+    /**
+     * The ids of every peer addPeers() selected, which is 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_t>
+    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};
+    std::optional<std::size_t> firstLimit_;
+    std::set<Peer::id_t> addedPeers_;
+};
+
+/**
+ * The given nodes of a chain as a TMLedgerData, so a test can go through the
+ * real dispatch rather than calling an acquisition directly.
+ *
+ * Not part of DeepChain itself: that header is shared with the gtest suites,
+ * whose binary has neither the protobuf types nor anything to send them to.
+ *
+ * @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 instead, 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, so sending the wrong one is rejected outright.
+        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, so a case cannot simply look once. The deadline
+ * is generous, so a loaded machine does not turn a pass into a failure.
+ *
+ * @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 rather than get(): that string is a log format, not an API. It is
+ * pinned once, in the SHAMapAddNode tests, and is what to pass BEAST_EXPECTS()
+ * as the reason a check here failed. Same name and meaning as the gtest suites'
+ * tallyIs(), which returns an AssertionResult instead.
+ *
+ * @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/InboundLedger_test.cpp

@@ -0,0 +1,378 @@
+#include <test/app/AcquireTestHelpers.h>
+#include <test/jtx/Env.h>
+
+#include <xrpld/app/ledger/InboundLedger.h>
+#include <xrpld/app/ledger/InboundLedgers.h>
+#include <xrpld/app/ledger/LedgerMaster.h>
+
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/basics/chrono.h>
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/nodestore/NodeObject.h>
+#include <xrpl/protocol/HashPrefix.h>
+#include <xrpl/protocol/LedgerHeader.h>
+#include <xrpl/protocol/Serializer.h>
+
+#include <chrono>
+#include <memory>
+#include <mutex>
+#include <set>
+#include <utility>
+#include <vector>
+
+namespace xrpl::test {
+
+/**
+ * An acquisition that exposes the entry points its bases keep protected, so a
+ * case can reach them without the daemon's API growing.
+ */
+struct TestableInboundLedger final : InboundLedger
+{
+    using InboundLedger::InboundLedger;
+
+    /**
+     * Look for the ledger locally, ask the peers being tracked for the
+     * rest, and arm the timer, as InboundLedgers::acquire() does.
+     *
+     * That caller holds its collection lock across init(), which releases
+     * it, so this stands in with a lock of its own. Declared before the
+     * lock, so it outlives it.
+     */
+    void
+    startAcquire()
+    {
+        std::recursive_mutex collectionMutex;
+        ScopedLockType collectionLock(collectionMutex);
+        init(collectionLock);
+    }
+
+    /**
+     * Ask for more nodes, or judge what has been collected, as a fresh
+     * acquisition does.
+     */
+    void
+    triggerAdded()
+    {
+        trigger(nullptr, TriggerReason::Added);
+    }
+};
+
+struct InboundLedger_test : public beast::unit_test::Suite
+{
+    /**
+     * A retry interval short enough that a whole timeout chain costs a fraction
+     * of a second. TimeoutCounter refuses anything at or below 10ms.
+     */
+    static constexpr auto kFastRetry = std::chrono::milliseconds{20};
+
+    /**
+     * A seed no other chain in this suite has used.
+     *
+     * The Env below is shared, and its node store, fetch packs and remembered
+     * failures are all keyed by hash, so two cases building identically seeded
+     * chains would let one resolve or judge the other's. Handing out a fresh
+     * seed per chain makes that impossible rather than merely unlikely.
+     *
+     * @return The seed.
+     */
+    [[nodiscard]] unsigned int
+    nextSeed()
+    {
+        return ++seed_;
+    }
+
+    /**
+     * A ledger header naming the given map roots.
+     *
+     * The hash is derived from the fields, so an acquisition accepts the
+     * header as its own however the roots are chosen.
+     *
+     * @param txHash The transaction map root; zero means no transactions.
+     * @param accountHash The state map root; zero is a ledger no
+     *        acquisition can finish.
+     * @return The header, with its hash filled in.
+     */
+    static LedgerHeader
+    makeHeader(uint256 const& txHash, uint256 const& accountHash)
+    {
+        LedgerHeader header;
+        header.seq = 2;
+        header.parentCloseTime = NetClock::time_point{};
+        header.closeTime = NetClock::time_point{};
+        header.closeTimeResolution = NetClock::duration{10};
+        header.closeFlags = 0;
+        header.txHash = txHash;
+        header.accountHash = accountHash;
+        header.hash = calculateLedgerHash(header);
+        return header;
+    }
+
+    /**
+     * The common shape: no transactions, so only the state map is in play.
+     *
+     * @param chain The chain whose root to name as the state hash.
+     * @return The header, with its hash filled in.
+     */
+    static LedgerHeader
+    makeHeader(DeepChain const& chain)
+    {
+        return makeHeader(uint256{}, chain.rootHash.asUInt256());
+    }
+
+    /**
+     * Put the header in the local store, which is the first place tryDB()
+     * looks.
+     *
+     * Unlike a fetch pack, which hands each entry out once, the store
+     * keeps it, so more than one acquisition of the same ledger can find
+     * it.
+     *
+     * @param env The environment whose node store to seed.
+     * @param header The header to store, keyed by its own hash.
+     */
+    static void
+    storeHeader(jtx::Env& env, LedgerHeader const& header)
+    {
+        Serializer s;
+        s.add32(HashPrefix::LedgerMaster);
+        addRaw(header, s);
+
+        env.app().getNodeFamily().db().store(
+            NodeObjectType::Ledger, std::move(s.modData()), header.hash, header.seq);
+    }
+
+    /**
+     * Put every node of a chain in the local store, so a state-map walk
+     * resolves the whole map without a peer.
+     *
+     * @param env The environment whose node store to seed.
+     * @param header The header whose sequence the nodes are stored under.
+     * @param chain The chain supplying the nodes.
+     * @param maxDepth The deepest node to store, so a caller can leave a walk
+     *        something to ask for.
+     */
+    static void
+    storeStateNodes(
+        jtx::Env& env,
+        LedgerHeader const& header,
+        DeepChain const& chain,
+        unsigned int maxDepth)
+    {
+        auto& db = env.app().getNodeFamily().db();
+
+        for (auto depth = 0u; depth <= maxDepth; ++depth)
+        {
+            db.store(
+                NodeObjectType::AccountNode,
+                chain.prefixedNodeAt(depth),
+                chain.nodeAt(depth)->getHash().asUInt256(),
+                header.seq);
+        }
+    }
+
+    /**
+     * A ledger whose maps all resolve locally finishes on the spot, and the
+     * finished ledger is immutable and handed on.
+     *
+     * The case where tryDB() alone completes the acquisition, so it covers
+     * tryDB() reporting a ledger it found and done() taking its success arm
+     * on that path. Both entry points are driven: InboundLedgers::acquire()
+     * is the only caller of init(), and hands back the finished ledger
+     * itself, while checkLocal() is the route that reaches done().
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testLocalLedgerCompletesAcquire(jtx::Env& env)
+    {
+        testcase("A ledger found locally completes the acquire");
+
+        // A chain ending in a real leaf, so the state map is genuinely complete rather than merely
+        // rooted. No transactions, so only the state map is in play.
+        auto const chain = DeepChain::toLeaf(2, nextSeed());
+        auto const header = makeHeader(chain);
+
+        storeHeader(env, header);
+        storeStateNodes(env, header, chain, chain.deepestDepth);
+
+        // acquire() runs init() under its own collection lock and returns the ledger only once the
+        // acquisition is complete and unfailed, so a non-null result is what shows tryDB() found it
+        // without a peer ever being asked.
+        auto const acquired = env.app().getInboundLedgers().acquire(
+            header.hash, header.seq, InboundLedger::Reason::GENERIC);
+
+        BEAST_EXPECT(acquired != nullptr);
+        if (acquired)
+        {
+            BEAST_EXPECT(acquired->isImmutable());
+            BEAST_EXPECT(acquired->header().hash == header.hash);
+        }
+
+        // init() hands a ledger it completed to LedgerMaster itself, which is what makes it
+        // available to everything else.
+        BEAST_EXPECT(env.app().getLedgerMaster().getLedgerByHash(header.hash) != nullptr);
+
+        // Nothing was logged as a failure, which is the other arm of done().
+        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
+
+        // The same ledger through checkLocal(), which unlike init() reaches done(). Everything it
+        // needs is still in the store, since the first acquisition read rather than consumed it.
+        auto again = std::make_shared<InboundLedger>(
+            env.app(),
+            header.hash,
+            header.seq,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::make_unique<RequestCountingPeerSet>());
+
+        // True because the acquisition ended, which here means it succeeded, and it reports that
+        // only after done() has run.
+        BEAST_EXPECT(again->checkLocal());
+        BEAST_EXPECT(again->isComplete());
+        BEAST_EXPECT(!again->isFailed());
+
+        auto const settled = again->getLedger();
+        BEAST_EXPECT(settled != nullptr);
+        if (settled)
+            BEAST_EXPECT(settled->isImmutable());
+
+        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
+    }
+
+    /**
+     * An acquisition that fails on local data must still signal.
+     *
+     * Both entry points that reach tryDB() are covered, since without
+     * done() the object never signals, logFailure() never runs, and the
+     * hash never lands in recentFailures_ - so the same doomed ledger is
+     * asked for again on the next round. recentFailures_ is what the
+     * assertions watch, since it is the caller-visible consequence of
+     * having signalled.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testLocalFailureSignalsDone(jtx::Env& env)
+    {
+        testcase("An acquisition that fails locally still signals");
+
+        // A zero account hash is a ledger no acquisition can ever finish, and tryDB() says so as
+        // soon as it has the header.
+        auto const header = makeHeader(uint256{}, uint256{});
+        storeHeader(env, header);
+
+        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
+
+        // acquire() is the only caller of init(), and hands back nothing for a failed acquisition.
+        BEAST_EXPECT(
+            env.app().getInboundLedgers().acquire(
+                header.hash, header.seq, InboundLedger::Reason::GENERIC) == nullptr);
+
+        // The failure reached recentFailures_, which is what stops the next round asking again.
+        BEAST_EXPECT(waitFor([&] { return env.app().getInboundLedgers().isFailure(header.hash); }));
+
+        // The other route into tryDB(): a trigger() on an acquisition that has no header yet. A
+        // hash of its own, so the entry above cannot answer for it.
+        auto const otherHeader = makeHeader(uint256{1}, uint256{});
+        storeHeader(env, otherHeader);
+
+        auto viaTrigger = std::make_shared<TestableInboundLedger>(
+            env.app(),
+            otherHeader.hash,
+            otherHeader.seq,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::make_unique<RequestCountingPeerSet>());
+
+        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(otherHeader.hash));
+
+        viaTrigger->triggerAdded();
+
+        BEAST_EXPECT(viaTrigger->isFailed());
+        BEAST_EXPECT(!viaTrigger->isComplete());
+        BEAST_EXPECT(
+            waitFor([&] { return env.app().getInboundLedgers().isFailure(otherHeader.hash); }));
+    }
+
+    /**
+     * The retry timer re-asks, then gives up and signals.
+     *
+     * The only case that drives onTimer() rather than trigger() directly,
+     * which is what covers the give-up: past kLedgerTimeoutRetriesMax the
+     * acquisition fails itself and done() records that, so the same
+     * doomed ledger is not asked for again on the next round. It is also
+     * what the retry interval is a constructor parameter for, since the
+     * chain runs past kLedgerTimeoutRetriesMax ticks of three seconds
+     * apiece in production.
+     *
+     * Nothing is local and no data ever arrives, so no tick can record progress
+     * and the count only climbs. A hash of its own, so no other case can have
+     * remembered it as a failure already.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testTimerRetriesThenGivesUp(jtx::Env& env)
+    {
+        testcase("The retry timer re-asks, then gives up");
+
+        uint256 const kUnknownLedger{8};
+
+        // One candidate, which onTimer() re-offers on every tick.
+        auto const candidate = std::make_shared<ChargeRecordingPeer>();
+        auto peerSet =
+            std::make_unique<RequestCountingPeerSet>(std::vector<std::shared_ptr<Peer>>{candidate});
+        auto* const peerSetPtr = peerSet.get();
+
+        auto acquire = std::make_shared<TestableInboundLedger>(
+            env.app(),
+            kUnknownLedger,
+            0,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::move(peerSet),
+            kFastRetry);
+
+        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(kUnknownLedger));
+
+        // init() finds nothing locally, so it asks the candidate and queues the first check-in,
+        // which is what arms the retry timer for every cycle after. Those first requests are not
+        // the ones under test, so count from here.
+        acquire->startAcquire();
+        BEAST_EXPECT(!acquire->isFailed());
+        int const requestsFromInit = peerSetPtr->requests();
+        BEAST_EXPECT(requestsFromInit > 0);
+        BEAST_EXPECT(peerSetPtr->addedPeers() == std::set<Peer::id_t>{candidate->id()});
+
+        // Every tick asks again, and past kLedgerTimeoutRetriesMax (6) the chain gives up.
+        BEAST_EXPECT(waitFor([&] { return acquire->isFailed(); }));
+        BEAST_EXPECT(!acquire->isComplete());
+        BEAST_EXPECT(peerSetPtr->requests() > requestsFromInit);
+
+        // done() remembered the hash, which is what stops the next round asking again.
+        BEAST_EXPECT(
+            waitFor([&] { return env.app().getInboundLedgers().isFailure(kUnknownLedger); }));
+    }
+
+    void
+    run() override
+    {
+        // One Env for the suite, since building one costs far more than any case here. Safe
+        // because every chain is seeded through nextSeed(): the node store, the fetch packs and
+        // the remembered failures are all shared, and all three are keyed by hash.
+        jtx::Env env{*this};
+
+        testLocalLedgerCompletesAcquire(env);
+        testLocalFailureSignalsDone(env);
+
+        // Last: the only case that waits out a whole timeout chain.
+        testTimerRetriesThenGivesUp(env);
+    }
+
+private:
+    unsigned int seed_{0};
+};
+
+BEAST_DEFINE_TESTSUITE(InboundLedger, app, xrpl);
+
+}  // namespace xrpl::test

src/test/app/TransactionAcquire_test.cpp

@@ -0,0 +1,512 @@
+#include <test/app/AcquireTestHelpers.h>
+#include <test/jtx/Env.h>
+
+#include <xrpld/app/ledger/InboundTransactions.h>
+#include <xrpld/app/ledger/detail/TransactionAcquire.h>
+#include <xrpld/overlay/Peer.h>
+
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/resource/Fees.h>
+#include <xrpl/shamap/SHAMap.h>
+#include <xrpl/shamap/SHAMapAddNode.h>
+#include <xrpl/shamap/SHAMapNodeID.h>
+
+#include <xrpl.pb.h>
+
+#include <chrono>
+#include <memory>
+#include <set>
+#include <utility>
+#include <vector>
+
+namespace xrpl::test {
+
+/**
+ * An acquisition that exposes the state its bases keep protected, so a case can
+ * reach it without the daemon's API growing.
+ */
+struct TestableTransactionAcquire final : TransactionAcquire
+{
+    using TransactionAcquire::TransactionAcquire;
+
+    /**
+     * Whether the set being acquired is still structurally coherent.
+     *
+     * @return Whether the map is still valid.
+     */
+    [[nodiscard]] bool
+    isMapValid() const
+    {
+        // Under the lock: a batch on another thread can reach the verdict.
+        ScopedLockType const sl(mtx_);
+        return map_->isValid();
+    }
+
+    /**
+     * Whether a batch has advanced the set since the flag was last cleared.
+     *
+     * @return Whether progress has been recorded.
+     */
+    [[nodiscard]] bool
+    madeProgress() const
+    {
+        // Under the lock: a timer tick clears the flag on a job thread.
+        ScopedLockType const sl(mtx_);
+        return progress_;
+    }
+
+    /**
+     * Forget any recorded progress.
+     */
+    void
+    clearProgress()
+    {
+        ScopedLockType const sl(mtx_);
+        progress_ = false;
+    }
+};
+
+struct TransactionAcquire_test : public beast::unit_test::Suite
+{
+    /**
+     * A retry interval short enough that a whole timeout chain costs a fraction
+     * of a second.
+     *
+     * TimeoutCounter refuses anything at or below 10ms. At this interval the
+     * window between the first retry (four timeouts in) and giving up (twenty)
+     * is still a third of a second, which is what the one case that watches
+     * both needs.
+     */
+    static constexpr auto kFastRetry = std::chrono::milliseconds{20};
+
+    /**
+     * A seed no other chain in this suite has used.
+     *
+     * The Env below is shared, and ConsensusTransSetSF::gotNode() puts
+     * every node it accepts into the application-wide NodeCache while
+     * InboundTransactions keys its acquisitions by set hash, so two cases
+     * building identically seeded chains would let one resolve or revive
+     * the other's. Handing out a fresh seed per chain makes that
+     * impossible rather than merely unlikely.
+     *
+     * @return The seed.
+     */
+    [[nodiscard]] unsigned int
+    nextSeed()
+    {
+        return ++seed_;
+    }
+
+    /**
+     * Whether takeNodes() declined to look at the data at all.
+     *
+     * TimeoutCounter::complete_ and failed_ are both protected, so this stands in
+     * for either: a done acquisition returns a verdict accounting for nothing.
+     *
+     * @param san The verdict a takeNodes() call returned.
+     * @return Whether that verdict shows the data was never looked at.
+     */
+    static bool
+    wasIgnored(SHAMapAddNode const& san)
+    {
+        return tallyIs(san, 0, 0, 0);
+    }
+
+    /**
+     * Wait for a finished set to reach InboundTransactions.
+     *
+     * done() hands the map over through a job, so this is what shows an
+     * acquisition completed rather than merely stopped.
+     *
+     * @param env The environment whose InboundTransactions to watch.
+     * @param setHash The set to wait for.
+     * @return The delivered map, or nullptr if none arrived.
+     */
+    [[nodiscard]] static std::shared_ptr<SHAMap>
+    waitForDeliveredSet(jtx::Env& env, uint256 const& setHash)
+    {
+        auto& inbound = env.app().getInboundTransactions();
+
+        // acquire=false: asking to acquire would spin up a second, unrelated acquisition for
+        // this hash (or, if one is already registered, needlessly poke stillNeed() on it).
+        //
+        // A shorter deadline than the polling default: the job is queued before takeNodes()
+        // returns, so anything beyond a few seconds means it is never coming.
+        std::shared_ptr<SHAMap> delivered;
+        if (!waitFor(
+                [&] { return (delivered = inbound.getSet(setHash, false)) != nullptr; },
+                std::chrono::seconds{5}))
+            return nullptr;
+        return delivered;
+    }
+
+    /**
+     * A chain ending in a real leaf completes the acquisition, and later
+     * replies for it are then left alone.
+     *
+     * Also pins the two things that follow from finishing: nothing more
+     * is asked for, and done() hands the map to InboundTransactions,
+     * which is what consensus is waiting on.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testHappyPathCompletesAcquisition(jtx::Env& env)
+    {
+        testcase("A chain ending in a leaf completes the acquire");
+
+        auto const chain = DeepChain::toLeaf(3, nextSeed());
+
+        auto peerSet = std::make_unique<RequestCountingPeerSet>();
+        auto* const peerSetPtr = peerSet.get();
+
+        uint256 const setHash = chain.rootHash.asUInt256();
+        auto const acquire =
+            std::make_shared<TransactionAcquire>(env.app(), setHash, std::move(peerSet));
+        auto const peer = std::make_shared<ChargeRecordingPeer>();
+
+        // The root alone leaves the set incomplete, so accepting it asks for the level
+        // below.
+        auto const rootResult = acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer);
+        BEAST_EXPECT(rootResult.isUseful());
+        int const requestsWhileIncomplete = peerSetPtr->requests();
+        BEAST_EXPECT(requestsWhileIncomplete > 0);
+
+        // The rest of the chain, ending in the leaf.
+        auto const result = acquire->takeNodes(chain.nodesBelowRoot(), peer);
+        BEAST_EXPECT(result.isUseful());
+        BEAST_EXPECT(!result.isInvalid());
+
+        // Nothing more went out, which rules out the acquisition still asking for nodes.
+        // trigger() also sends nothing when it gives up, so the delivered set below is what
+        // shows it finished.
+        BEAST_EXPECT(peerSetPtr->requests() == requestsWhileIncomplete);
+
+        // done() hands the map over only when it has not failed, so this is what separates
+        // completion from failure.
+        auto const delivered = waitForDeliveredSet(env, setHash);
+        BEAST_EXPECT(delivered != nullptr);
+        if (delivered)
+        {
+            BEAST_EXPECT(delivered->getHash() == chain.rootHash);
+            BEAST_EXPECT(delivered->isValid());
+        }
+
+        // A reply arriving after the set is finished is not examined, and does not restart the
+        // asking - the ordinary fate of every responder to trigger()'s broadcast but the one that
+        // completed the set. init() was never called, so no timer can have failed the acquisition
+        // since the set above was delivered.
+        BEAST_EXPECT(wasIgnored(acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer)));
+        BEAST_EXPECT(peerSetPtr->requests() == requestsWhileIncomplete);
+    }
+
+    /**
+     * Two peers each answering with a different missing piece are both accepted
+     * without penalty, and the set completes from their combined replies.
+     *
+     * Driven through InboundTransactions::gotData() so the leaf goes over the
+     * real dispatch, which rebuilds a leaf's position from its own key rather
+     * than trusting the sender's label.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testTwoPeersEachSupplyPartOfTheSet(jtx::Env& env)
+    {
+        testcase("Two peers each supplying part of a set are both accepted without penalty");
+
+        auto const chain = DeepChain::toLeaf(3, nextSeed());
+        auto& inbound = env.app().getInboundTransactions();
+
+        // getSet() with acquire=true registers the TransactionAcquire that gotData() then
+        // looks up by hash.
+        uint256 const setHash = chain.rootHash.asUInt256();
+        BEAST_EXPECT(inbound.getSet(setHash, true) == nullptr);
+
+        // The first peer answers with the root only.
+        auto const peerA = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, peerA, packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}}));
+        BEAST_EXPECT(peerA->charges().empty());
+
+        // The second answers with everything the first left out, and finishes the set.
+        auto const peerB = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, peerB, packetFor(chain, chain.nodesBelowRoot()));
+        BEAST_EXPECT(peerB->charges().empty());
+
+        // Completion does not depend on which peer sent which piece: the set finishes just
+        // as it does when one peer supplies the lot.
+        auto const delivered = waitForDeliveredSet(env, setHash);
+        BEAST_EXPECT(delivered != nullptr);
+        if (delivered)
+            BEAST_EXPECT(delivered->getHash() == chain.rootHash);
+    }
+
+    /**
+     * A root that does not hash to the set we asked for is a plain mismatch,
+     * and has to leave the acquisition able to try another peer.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testBadRootKeepsAcquireAlive(jtx::Env& env)
+    {
+        testcase("A mismatched root leaves the acquire recoverable");
+
+        DeepChain const chain{nextSeed()};
+
+        // Acquire an unrelated hash, so the chain's root cannot match it.
+        auto const acquire = std::make_shared<TestableTransactionAcquire>(
+            env.app(), uint256{42}, std::make_unique<RequestCountingPeerSet>());
+        auto const peer = std::make_shared<ChargeRecordingPeer>();
+
+        auto const result = acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer);
+        BEAST_EXPECTS(tallyIs(result, 0, 1, 0), result.get());
+
+        // A mismatched root says nothing about the tree behind the hash we asked for, so the map
+        // is untouched.
+        BEAST_EXPECT(acquire->isMapValid());
+
+        // Still alive: the next packet is examined rather than waved through.
+        BEAST_EXPECT(!wasIgnored(acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer)));
+    }
+
+    /**
+     * A second reply carrying a root we already have must stay free.
+     *
+     * This is what an honest second responder to the initial fan-out sends:
+     * trigger() broadcasts to every tracked peer, so several answer the same
+     * request and all but the first carry nothing new. Charging for that would
+     * penalize peers for answering.
+     *
+     * Covers the root specifically, which takeNodes() short-circuits on
+     * haveRoot_ without consulting the map. A repeated non-root node takes the
+     * other route, through addKnownNode() - see
+     * testDuplicateNonRootReplyIsFree().
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testDuplicateRootReplyIsFree(jtx::Env& env)
+    {
+        testcase("A reply of a root we already have is free");
+
+        DeepChain const chain{nextSeed()};
+        auto& inbound = env.app().getInboundTransactions();
+
+        // getSet() with acquire=true registers the TransactionAcquire that gotData() then looks up
+        // by hash.
+        uint256 const setHash = chain.rootHash.asUInt256();
+        BEAST_EXPECT(inbound.getSet(setHash, true) == nullptr);
+
+        auto const rootPacket = packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}});
+
+        // The first responder supplies the root, which is genuinely useful.
+        auto const firstPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, firstPeer, rootPacket);
+        BEAST_EXPECT(firstPeer->charges().empty());
+
+        // The second sends the same root. Nothing is added to the map, but the peer did what we
+        // asked, so it must not be charged.
+        auto const secondPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, secondPeer, rootPacket);
+        BEAST_EXPECT(secondPeer->charges().empty());
+    }
+
+    /**
+     * A repeated non-root node must stay free too.
+     *
+     * The counterpart to testDuplicateRootReplyIsFree(), covering the
+     * route that does consult the map: addKnownNode() reports a node it
+     * already holds as a duplicate, and takeNodes() tests isGood(), which
+     * counts a duplicate as success. Testing isUseful() there instead
+     * would turn every honest second responder into a peer we charge for
+     * invalid data.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testDuplicateNonRootReplyIsFree(jtx::Env& env)
+    {
+        testcase("A repeated non-root node is free");
+
+        auto const chain = DeepChain::toLeaf(3, nextSeed());
+        auto& inbound = env.app().getInboundTransactions();
+
+        uint256 const setHash = chain.rootHash.asUInt256();
+        BEAST_EXPECT(inbound.getSet(setHash, true) == nullptr);
+
+        auto const rootPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, rootPeer, packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}}));
+        BEAST_EXPECT(rootPeer->charges().empty());
+
+        // Depth 1 alone, so the set stays incomplete and the acquisition keeps examining data
+        // rather than waving the second copy through as a late reply.
+        auto const level1 = packetFor(chain, {{chain.idAt(1), chain.nodeAt(1)}});
+
+        auto const firstPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, firstPeer, level1);
+        BEAST_EXPECT(firstPeer->charges().empty());
+
+        auto const secondPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, secondPeer, level1);
+        BEAST_EXPECT(secondPeer->charges().empty());
+    }
+
+    /**
+     * A reply whose node data cannot be deserialized is charged for.
+     *
+     * gotData() rejects the packet before the acquisition is handed
+     * anything, so this pins the charge on the dispatch layer rather than
+     * on takeNodes(). It also gives this suite's "was not charged"
+     * assertions their teeth: a harness that recorded no charge at all
+     * would satisfy all of them and fail only here.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testUndeserializableNodeIsCharged(jtx::Env& env)
+    {
+        testcase("A reply with undeserializable node data is charged");
+
+        DeepChain const chain{nextSeed()};
+        auto& inbound = env.app().getInboundTransactions();
+
+        uint256 const setHash = chain.rootHash.asUInt256();
+        BEAST_EXPECT(inbound.getSet(setHash, true) == nullptr);
+
+        // A single byte naming a wire type that does not exist, so getTreeNode() rejects it
+        // before the acquisition is handed anything.
+        auto packet = std::make_shared<protocol::TMLedgerData>();
+        packet->set_ledgerhash(setHash.data(), uint256::size());
+        packet->set_ledgerseq(0);
+        packet->set_type(protocol::liTS_CANDIDATE);
+
+        auto* const node = packet->add_nodes();
+        node->set_nodedata("\xff", 1);
+        node->set_id(SHAMapNodeID{}.getRawString());
+
+        auto const peer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, peer, packet);
+
+        BEAST_EXPECT(peer->charges() == std::vector{resource::kFeeInvalidData});
+    }
+
+    /**
+     * init() asks only the peers that claim to have the set.
+     *
+     * addPeers() passes hasTxSet(hash_) as its filter and trigger() as its
+     * callback, so a peer that says it has the set is asked and one that says
+     * it does not is left alone. Getting this wrong wastes a request on every
+     * peer in the overlay for every set.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testInitAsksOnlyPeersWithTheSet(jtx::Env& env)
+    {
+        testcase("init() asks only the peers that have the set");
+
+        DeepChain const chain{nextSeed()};
+
+        // Ordered with the useless peer first, so a filter that is ignored altogether
+        // shows up as the wrong peer being asked rather than as one extra request.
+        auto const withoutSet = std::make_shared<ChargeRecordingPeer>(false);
+        auto const withSet = std::make_shared<ChargeRecordingPeer>(true);
+
+        auto peerSet = std::make_unique<RequestCountingPeerSet>(
+            std::vector<std::shared_ptr<Peer>>{withoutSet, withSet});
+        auto* const peerSetPtr = peerSet.get();
+
+        auto const acquire = std::make_shared<TransactionAcquire>(
+            env.app(), chain.rootHash.asUInt256(), std::move(peerSet));
+
+        static constexpr int kStartPeers = 2;
+        acquire->init(kStartPeers);
+
+        // Stop the retry loop, which would otherwise keep offering the same candidates
+        // for as long as this case runs.
+        acquire->cancel();
+
+        BEAST_EXPECT(peerSetPtr->firstLimit() == kStartPeers);
+        BEAST_EXPECT(peerSetPtr->addedPeers() == std::set<Peer::id_t>{withSet->id()});
+
+        // The peer that was added is also asked, rather than merely tracked.
+        BEAST_EXPECT(peerSetPtr->requests() >= 1);
+    }
+
+    /**
+     * The retry timer re-asks with no peer of its own, then gives up on
+     * its own.
+     *
+     * The only case that reaches onTimer(). Pins the two behaviors, not the
+     * thresholds they trip at: bounding those means asserting on wall clock.
+     * Both are read in one poll, so a fast interval cannot let the give-up land
+     * between the two readings.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testTimerRetriesThenGivesUp(jtx::Env& env)
+    {
+        testcase("The retry timer re-asks, then gives up");
+
+        DeepChain const chain{nextSeed()};
+
+        auto peerSet = std::make_unique<RequestCountingPeerSet>();
+        auto* const peerSetPtr = peerSet.get();
+
+        // An unrelated hash, so the probe below can never be accepted. See probe().
+        auto const acquire = std::make_shared<TransactionAcquire>(
+            env.app(), uint256{42}, std::move(peerSet), kFastRetry);
+
+        // No candidates, so nothing goes out until onTimer() decides to broadcast.
+        acquire->init(1);
+        BEAST_EXPECT(peerSetPtr->requests() == 0);
+
+        // A root that cannot hash to this acquisition's set is rejected without recording
+        // progress, so polling with it does not postpone the timeout being waited for. A fresh
+        // peer each time keeps the rejections from piling up on one.
+        auto const probe = [&] {
+            return wasIgnored(acquire->takeNodes(
+                {{SHAMapNodeID{}, chain.nodeAt(0)}}, std::make_shared<ChargeRecordingPeer>()));
+        };
+
+        // kNormTimeouts (4) intervals in, onTimer() starts asking again with no peer of its own to
+        // ask, and the acquisition is still examining data at that point.
+        BEAST_EXPECT(waitFor([&] { return peerSetPtr->requests() > 0 && !probe(); }));
+
+        // Past kMaxTimeouts (20) it fails itself, and stops examining data.
+        BEAST_EXPECT(waitFor(probe));
+
+        // Whichever poll saw the retry, the count it left behind is what records that it happened.
+        BEAST_EXPECT(peerSetPtr->requests() > 0);
+    }
+
+    void
+    run() override
+    {
+        // One Env for the suite, since building one costs far more than any case here. Safe
+        // because every chain is seeded through nextSeed(): gotNode() puts every node it accepts
+        // into the application-wide NodeCache, so cases sharing an Env must not share a hash.
+        jtx::Env env{*this};
+
+        testHappyPathCompletesAcquisition(env);
+        testTwoPeersEachSupplyPartOfTheSet(env);
+        testBadRootKeepsAcquireAlive(env);
+        testDuplicateRootReplyIsFree(env);
+        testDuplicateNonRootReplyIsFree(env);
+        testUndeserializableNodeIsCharged(env);
+        testInitAsksOnlyPeersWithTheSet(env);
+
+        // Last: the only case that waits out a whole timeout chain.
+        testTimerRetriesThenGivesUp(env);
+    }
+
+private:
+    unsigned int seed_{0};
+};
+
+BEAST_DEFINE_TESTSUITE(TransactionAcquire, app, xrpl);
+
+}  // namespace xrpl::test

src/tests/libxrpl/shamap/DeepChain.h

@@ -0,0 +1,304 @@
+#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, each with one real child (and, when built with a decoy, an
+ * additional unresolvable second child), from the root down to one deepest node, in
+ * wire form.
+ *
+ * Built bottom-up, so the root hash commits to the whole shape and every node
+ * hashes correctly, which lets a test supply one level at a time and have each
+ * accepted on its own merits. Every node sits on the branch pathKey selects at
+ * its depth, so a receiver descending towards that key walks the whole chain.
+ *
+ * Two shapes, separating a chain a peer could honestly send from one it could
+ * not:
+ *
+ * - the constructors run inner nodes all the way to SHAMap::kLeafDepth,
+ *   where no valid tree can hold one (see the badDepth check in
+ *   SHAMap::addKnownNode), so feeding it invalidates the map;
+ * - toLeaf() stops at a real transaction leaf, so feeding it completes an
+ *   acquisition.
+ *
+ * Shared by the gtest suites and the daemon's own, which is why nothing here
+ * reaches outside libxrpl: xrpl_tests links only xrpl.libxrpl, while Peer,
+ * PeerSet and the protobuf reply builder are all xrpld. The xrpld half of this
+ * helper 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
+    // fabricated chain, which therefore sits on branch 0 at every depth.
+    uint256 pathKey;
+
+    // The depth of the deepest node, which is 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. Published so a caller can relate it to a threshold of
+     * its own, as AcquireTestHelpers.h does.
+     */
+    static constexpr std::size_t kLeafItemBytes = kMinShaMapItemBytes;
+
+    /**
+     * A chain of inner nodes reaching SHAMap::kLeafDepth, which no valid tree
+     * can hold.
+     *
+     * @param seed Varies the whole chain, so two chains can coexist without one
+     *        resolving the other's nodes. Caches and fetch packs are keyed by
+     *        hash, so identically-seeded chains are the same chain.
+     */
+    explicit DeepChain(unsigned int seed = 1) : DeepChain(std::nullopt, seed, Decoy::No)
+    {
+    }
+
+    /**
+     * 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. Deeper is sparser than a one-transaction set
+     *        would really be, but the sync path judges nodes by their hashes
+     *        rather than by how sparse they are.
+     * @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)
+    {
+        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;
+    }
+
+    /**
+     * Fill a synching map, stopping one level short of the deepest node so the
+     * caller offers that one itself.
+     *
+     * Reports rather than asserts, since this header is shared with a binary
+     * that has no test framework of its own to assert through.
+     *
+     * @param map The map to fill.
+     * @return Whether the root and every node above the deepest one was
+     *         accepted - a property of the chain, not of the map.
+     */
+    [[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 that no valid tree can hold.
+     *
+     * @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 is never stored anywhere.
+    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)
+        {
+            // No leaf, so the deepest inner node points at a child that is never fetched.
+            buildInnersDownTo(SHAMap::kLeafDepth, SHAMapHash{uint256{seed}}, decoy);
+            return;
+        }
+
+        // Exactly kLeafItemBytes of payload, the smallest a leaf item may be, which keeps the leaf
+        // below the size at which a receiver tries to parse one as a transaction. 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 the chain above it
+        // has no say in where it sits: it has to follow this key's nibbles.
+        pathKey = leafKey(*leaf);
+        deepestDepth = *leafDepth;
+        nodes[*leafDepth] = leaf;
+
+        if (*leafDepth == 0)
+        {
+            // The leaf is the root, so there is nothing above it to build.
+            rootHash = leaf->getHash();
+            return;
+        }
+
+        buildInnersDownTo(*leafDepth - 1, leaf->getHash(), decoy);
+    }
+
+    /**
+     * Fill in inner nodes, each with one real child (and, under Decoy::Yes, an
+     * additional unresolvable second child), from the root down to the given depth,
+     * and record the root hash.
+     *
+     * 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, which is the fabricated chain's whole point.
+     * @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)
+    {
+        for (auto depth = deepest + 1; depth-- > 0;)
+        {
+            // A key has only 64 nibbles, so selectBranch() at SHAMap::kLeafDepth would index one
+            // byte past the end of the 32-byte key. A fabricated chain's pathKey is zero, so
+            // branch 0 is the position such a node claims anyway.
+            auto const branch =
+                depth == SHAMap::kLeafDepth ? 0u : selectBranch(idAt(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 only stays free of the real child because
+                // every caller that passes Decoy::Yes leaves pathKey at its default of zero. If a
+                // caller ever combined a non-zero pathKey with a decoy, the compressed-inner-node
+                // parser would silently let the decoy overwrite the real child's hash instead of
+                // rejecting the duplicate branch, so guard the assumption rather than rely on it.
+                if (branch == 1)
+                    Throw<std::logic_error>("DeepChain: decoy branch collides with real child");
+
+                // Derived from the depth so it differs per level - each posts its own read - and
+                // is deterministic and cannot collide with a real node hash.
+                uint256 decoyHash;
+                decoyHash.begin()[0] = 0xDE;
+                decoyHash.begin()[1] = 0xC0;
+                decoyHash.begin()[2] = static_cast<unsigned char>(depth);
+                s.addBitString(decoyHash);
+                s.add8(1);  // the unresolvable decoy sits at branch 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/SHAMapAddNode.cpp

@@ -0,0 +1,81 @@
+#include <xrpl/shamap/SHAMapAddNode.h>
+
+#include <gtest/gtest.h>
+
+namespace xrpl::tests {
+
+// get() is a log format rather than an API, so it is pinned here, once, instead of at every site
+// that has a verdict to check. Those check the tally through getGood()/getBad()/getDuplicate()
+// (see tallyIs() in AcquireTestHelpers.h and SHAMapSync.cpp) or through
+// isGood()/isUseful()/isInvalid(), which say the same thing without depending on the wording.
+TEST(SHAMapAddNode, getNamesEveryNonEmptyCount)
+{
+    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 the tests assert on, and the verdicts derived from them, so a tally check and
+// the log line cannot drift apart.
+TEST(SHAMapAddNode, countsAndVerdictsAgree)
+{
+    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 was hooked in, 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 towards good without needing to be useful itself: isUseful() here still
+    // reflects the incUseful() above, not this increment.
+    san.incDuplicate();
+    EXPECT_EQ(san.getDuplicate(), 1);
+    EXPECT_FALSE(san.isInvalid());
+    EXPECT_TRUE(san.isGood());
+
+    // Bad is counted, not merely flagged: a batch that carries on past a rejected node reports one
+    // per node, so a test can tell "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";
+
+    // Adding one verdict to another sums every count, which is how a batch's verdict is built up
+    // one node at a time.
+    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/SHAMapSync.cpp

@@ -1,13 +1,20 @@
+#include <xrpl/basics/Blob.h>
 #include <xrpl/basics/SHAMapHash.h>
 #include <xrpl/basics/Slice.h>
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/basics/random.h>
+#include <xrpl/beast/hash/uhash.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/beast/xor_shift_engine.h>
+#include <xrpl/ledger/Ledger.h>
+#include <xrpl/protocol/LedgerHeader.h>
+#include <xrpl/protocol/Rules.h>
 #include <xrpl/protocol/Serializer.h>
 #include <xrpl/shamap/SHAMap.h>
 #include <xrpl/shamap/SHAMapItem.h>
 #include <xrpl/shamap/SHAMapMissingNode.h>
+#include <xrpl/shamap/SHAMapNodeID.h>
+#include <xrpl/shamap/SHAMapSyncFilter.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
 #include <boost/smart_ptr/intrusive_ptr.hpp>
@@ -20,11 +27,28 @@
 #include <cstddef>
 #include <cstdint>
 #include <list>
+#include <map>
+#include <optional>
+#include <unordered_set>
 #include <utility>
 #include <vector>
 
 namespace xrpl::tests {
 
+// The cap on how many nodes a walk reports, set well above what any test here expects.
+static constexpr int kMaxNodesPerRequest = 2048;
+
+/**
+ * Rules with no amendments enabled, which is all a ledger built here needs.
+ *
+ * @return The rules.
+ */
+[[nodiscard]] static Rules
+noAmendments()
+{
+    return Rules{std::unordered_set<uint256, beast::Uhash<>>{}};
+}
+
 class SHAMapSyncTest : public ::testing::Test
 {
 protected:
@@ -81,8 +105,241 @@ protected:
 
         return true;
     }
+
+    /**
+     * A single-child inner node in wire form, pointing at the given child.
+     *
+     * The chain a caller builds from these is put together bottom-up,
+     * since each node's hash covers the child hash below it.
+     *
+     * @param childHash What the node points at, on branch 0.
+     * @return The deserialized node.
+     */
+    [[nodiscard]] static SHAMapTreeNodePtr
+    makeInnerNode(SHAMapHash const& childHash)
+    {
+        Serializer s;
+        s.addBitString(childHash.asUInt256());
+        s.add8(0);  // the chain continues at branch 0
+        s.add8(kWireTypeCompressedInner);
+
+        return SHAMapTreeNode::makeFromWire(makeSlice(s.peekData()));
+    }
+
+    /**
+     * A sync filter that records every node it is told about, and serves back
+     * only the ones it was explicitly asked to hold.
+     *
+     * Serving is opt-in: the sync path consults the filter before deciding
+     * a node is missing, so a filter that served everything it had seen
+     * would resolve the node a test is about to offer.
+     */
+    class RecordingFilter : public SHAMapSyncFilter
+    {
+    public:
+        // What one gotNode() call was told, in the order the calls arrived.
+        struct Report
+        {
+            bool fromFilter;
+            SHAMapHash hash;
+            std::uint32_t ledgerSeq;
+        };
+
+        void
+        gotNode(
+            bool fromFilter,
+            SHAMapHash const& hash,
+            std::uint32_t ledgerSeq,
+            Blob&&,  // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved)
+            SHAMapNodeType) const override
+        {
+            reports_.push_back({.fromFilter = fromFilter, .hash = hash, .ledgerSeq = ledgerSeq});
+        }
+
+        [[nodiscard]] std::optional<Blob>
+        getNode(SHAMapHash const& hash) const override
+        {
+            if (auto const it = served_.find(hash); it != served_.end())
+                return it->second;
+            return std::nullopt;
+        }
+
+        /**
+         * Offer a node back to the map, the way a fetch pack would.
+         *
+         * @param node The node to serve, keyed by its own hash.
+         */
+        void
+        serve(SHAMapTreeNodePtr const& node)
+        {
+            Serializer s;
+            node->serializeWithPrefix(s);
+            served_.emplace(node->getHash(), s.modData());
+        }
+
+        [[nodiscard]] std::vector<Report> const&
+        reports() const
+        {
+            return reports_;
+        }
+
+    private:
+        // Mutable because the whole interface is const: a filter is handed to the map by
+        // const pointer, so recording has to happen through one.
+        mutable std::vector<Report> reports_;
+        std::map<SHAMapHash, Blob> served_;
+    };
+
+    /**
+     * A root inner node with all 16 branches occupied and not one of them
+     * resolvable.
+     *
+     * A walk of a backed map posts an asynchronous read for every branch in a
+     * single pass, so the nodestore reader threads run finishFetch() for the
+     * same map at the same time.
+     */
+    struct WideRoot
+    {
+        SHAMapTreeNodePtr node;
+        SHAMapHash hash;
+
+        WideRoot()
+        {
+            Serializer s;
+
+            for (auto branch = 0u; branch < SHAMap::kBranchFactor; ++branch)
+            {
+                // Derived from the branch so each posts its own read, and deterministic so it
+                // cannot collide with a real node hash.
+                uint256 childHash;
+                childHash.begin()[0] = 0xFA;
+                childHash.begin()[1] = 0xB1;
+                childHash.begin()[2] = static_cast<unsigned char>(branch);
+                s.addBitString(childHash);
+            }
+
+            s.add8(kWireTypeInner);
+
+            node = SHAMapTreeNode::makeFromWire(makeSlice(s.peekData()));
+            hash = node->getHash();
+        }
+    };
 };
 
+// A map marked complete in the database withdraws that claim the first time a read misses, and
+// reports the miss once so the ledger can be re-acquired. Sixteen unresolvable branches are posted
+// in one pass, so with four reader threads the misses overlap and finishFetch() runs concurrently
+// for a single map. The one-report count below is too narrow a window for an ordinary run to
+// police; what confirms it is the absence of a data race under ThreadSanitizer.
+TEST_F(SHAMapSyncTest, fullFlagIsWithdrawnOnceByConcurrentReaders)
+{
+    static constexpr auto kRounds = 8uz;
+    static constexpr auto kReadThreads = 4;
+
+    for (auto round = 0uz; round < kRounds; ++round)
+    {
+        TestNodeFamily f{j_, kReadThreads};
+        WideRoot const root;
+
+        // Backed, so descendAsync() posts real asynchronous reads rather than resolving inline.
+        SHAMap map{SHAMapType::FREE, f};
+        map.setSynching();
+
+        ASSERT_TRUE(map.addRootNode(root.hash, root.node, nullptr).isGood());
+
+        // The claim the first miss has to withdraw.
+        map.setFull();
+
+        // No filter, so every branch has to be read from a database that does not hold it.
+        EXPECT_EQ(map.getMissingNodes(kMaxNodesPerRequest, nullptr).size(), SHAMap::kBranchFactor)
+            << "round " << round;
+
+        EXPECT_EQ(f.missingBySeqReports(), 1uz) << "round " << round;
+    }
+}
+
+// Every node the sync path hands to a filter carries the map's ledger sequence, which is the hint
+// the filter passes on to a nodestore keyed by hash. All three call sites are covered: a root taken
+// from a peer, a node taken from a peer, and a node the walk resolved out of the filter itself.
+TEST_F(SHAMapSyncTest, syncFilterIsToldTheLedgerSequence)
+{
+    static constexpr std::uint32_t kLedgerSeq = 7;
+
+    TestNodeFamily f{j_};
+
+    // A three-level chain, built bottom-up so each hash covers the one below it. The deepest node
+    // points at a child that is never supplied, so the walk always has something to ask for.
+    auto const deepest = makeInnerNode(SHAMapHash{uint256{1}});
+    auto const middle = makeInnerNode(deepest->getHash());
+    auto const root = makeInnerNode(middle->getHash());
+
+    // Unbacked, so a node the filter does not hold is missing rather than looked up in a database.
+    SHAMap map{SHAMapType::FREE, f};
+    map.setUnbacked();
+    map.setSynching();
+    map.setLedgerSeq(kLedgerSeq);
+
+    RecordingFilter filter;
+
+    ASSERT_TRUE(map.addRootNode(root->getHash(), root, &filter).isGood());
+    ASSERT_TRUE(map.addKnownNode(SHAMapNodeID{1, uint256{}}, middle, &filter).isUseful());
+
+    // Only now is the deepest node resolvable, so nothing above it came out of the filter.
+    filter.serve(deepest);
+    auto const missing = map.getMissingNodes(kMaxNodesPerRequest, &filter);
+
+    // The walk resolved the deepest node through the filter and then asked for its child.
+    ASSERT_EQ(missing.size(), 1u);
+    EXPECT_EQ(missing[0].second, uint256{1});
+
+    ASSERT_EQ(filter.reports().size(), 3u);
+
+    // The two nodes taken from a peer, which the filter is told about so it can store them.
+    EXPECT_FALSE(filter.reports()[0].fromFilter);
+    EXPECT_EQ(filter.reports()[0].hash, root->getHash());
+    EXPECT_FALSE(filter.reports()[1].fromFilter);
+    EXPECT_EQ(filter.reports()[1].hash, middle->getHash());
+
+    // The one the walk read back out of the filter, which is reported as such.
+    EXPECT_TRUE(filter.reports()[2].fromFilter);
+    EXPECT_EQ(filter.reports()[2].hash, deepest->getHash());
+
+    for (auto const& report : filter.reports())
+        EXPECT_EQ(report.ledgerSeq, kLedgerSeq) << "hash " << report.hash;
+}
+
+// Ledger::setFull() has to publish each map's ledger sequence alongside the flag that lets the
+// first nodestore miss report a gap, or the report names the zero a map starts with and the lookup
+// it asks for cannot resolve.
+TEST_F(SHAMapSyncTest, ledgerSetFullPublishesTheLedgerSequence)
+{
+    static constexpr std::uint32_t kLedgerSeq = 7;
+
+    TestNodeFamily f{j_};
+
+    LedgerHeader header;
+    header.seq = kLedgerSeq;
+    // Non-zero, so the map has a root to look for and the lookup can miss.
+    header.txHash = uint256{1};
+    header.hash = calculateLedgerHash(header);
+
+    Ledger ledger{header, noAmendments(), f};
+
+    // The constructor already looked for that root and did not find it, but nothing had claimed the
+    // map was complete yet, so there was no gap to report.
+    ASSERT_EQ(f.missingBySeqReports(), 0uz);
+
+    ledger.setFull();
+
+    // Still missing, and now the map has a claim to withdraw, so the gap is reported.
+    // TestNodeFamily throws where the real family would start re-acquiring, which finishFetch()
+    // logs and swallows.
+    EXPECT_FALSE(ledger.txMap().fetchRoot(SHAMapHash{header.txHash}, nullptr));
+
+    EXPECT_EQ(f.missingBySeqReports(), 1uz);
+    EXPECT_EQ(f.missingBySeqRefNum(), kLedgerSeq);
+}
+
 TEST_F(SHAMapSyncTest, sync)
 {
     TestNodeFamily f{j_}, f2{j_};
@@ -92,7 +349,6 @@ TEST_F(SHAMapSyncTest, sync)
     static constexpr auto kItemCount = 10000uz;
     static constexpr auto kInvariantInterval = 100uz;
     static constexpr auto kNodesToConfuse = 500uz;
-    static constexpr auto kMaxNodesPerRequest = 2048;
 
     for (auto i = 0uz; i < kItemCount; ++i)
     {

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>
@@ -34,8 +36,18 @@ private:
 
     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. More than one is needed only by a test that wants
+     *        several reads to complete on different threads at once.
+     */
+    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 +62,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 +102,29 @@ 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 so a
+     *        test can check which sequence the map published.
+     * @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, as
+     * no test in this suite drives this path.
+     *
+     * @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,31 @@ 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, so a test that wants the
+     * count for one map has to give that map a family of its own.
+     *
+     * @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, which is the hint
+     * the map published for the nodestore lookup that has to resolve the gap.
+     *
+     * @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,18 @@ 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.
+     *
+     * A threshold rather than a derived bound: the smallest a hash-prefixed
+     * SHAMap leaf can be is the 4-byte HashPrefix plus kMinShaMapItemBytes, and
+     * nothing that size is a signed transaction either. The extra byte is the
+     * long-standing threshold this check has always used, kept as it was.
+     */
+    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 clock_type = beast::AbstractClock<std::chrono::steady_clock>;
@@ -44,13 +44,33 @@ public:
         CONSENSUS  // We believe the consensus round requires this ledger
     };
 
+    /**
+     * How long to wait between retries, and so how long each timeout counted
+     * against the acquisition takes. Long, since a ledger is worth chasing for
+     * far longer than a consensus round.
+     */
+    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. Defaulted in
+     *        production; InboundLedger_test passes a short one so a whole
+     *        timeout chain runs in a fraction of the time. TimeoutCounter
+     *        requires more than 10ms.
+     */
     InboundLedger(
         Application& app,
         uint256 const& hash,
         std::uint32_t seq,
         Reason reason,
-        clock_type&,
-        std::unique_ptr<PeerSet> peerSet);
+        clock_type& clock,
+        std::unique_ptr<PeerSet> peerSet,
+        std::chrono::milliseconds retryInterval = kRetryInterval);
 
     ~InboundLedger() override;
 
@@ -119,15 +139,35 @@ public:
         return lastAction_;
     }
 
-private:
+protected:
+    // Kept protected, with the two entry points naming it, so a test subclass (see
+    // InboundLedger_test) can drive an acquisition the way the timer chain does, without routing
+    // through the JobQueue. Production callers reach an acquisition through InboundLedgers.
+
+    // Why trigger() is being run, which decides how deep a request goes and whether the
+    // aggressive-retry branch is eligible.
     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 and signal whatever is waiting on it. Runs at most
+     * once. Call under mtx_, which the flags written here require.
+     */
+    void
+    done();
+
+private:
     void
     filterNodes(std::vector<std::pair<SHAMapNodeID, uint256>>& nodes, TriggerReason reason);
 
-    void
-    trigger(std::shared_ptr<Peer> const&, TriggerReason);
-
     std::vector<neededHash_t>
     getNeededHashes();
 
@@ -138,10 +178,7 @@ private:
     tryDB(node_store::Database& srcDB);
 
     void
-    done();
-
-    void
-    onTimer(bool progress, ScopedLockType& peerSetLock) override;
+    onTimer(bool progress, ScopedLockType& sl) override;
 
     std::size_t
     getPeerCount() const;

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,
     clock_type& 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,13 @@ InboundLedger::init(ScopedLockType& collectionLock)
     collectionLock.unlock();
 
     tryDB(app_.getNodeFamily().db());
+    // Matches checkLocal(): without done() this object never signals, so logFailure() never runs
+    // and the hash never lands in recentFailures_.
     if (failed_)
+    {
+        done();
         return;
+    }
 
     if (!complete_)
     {
@@ -497,6 +498,8 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         if (failed_)
         {
             JLOG(journal_.warn()) << " failed local for " << hash_;
+            // See init() for why done() must be called here rather than just returning.
+            done();
             return;
         }
     }

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

@@ -98,9 +98,16 @@ protected:
 
     /**
      * Hook called from invokeOnTimer().
+     *
+     * @param progress Whether the subtype recorded progress since the
+     *        last call.
+     * @param sl Proof mtx_ is held, and held for the whole call. It is
+     *        this object's own mutex rather than anything belonging to a
+     *        PeerSet, and it is recursive, so a nested lock taken inside
+     *        this call releases nothing when it goes out of scope.
      */
     virtual void
-    onTimer(bool progress, ScopedLockType&) = 0;
+    onTimer(bool progress, ScopedLockType& sl) = 0;
 
     /**
      * Return a weak pointer to this.

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

@@ -18,6 +18,7 @@
 #include <xrpl.pb.h>
 
 #include <algorithm>
+#include <chrono>
 #include <cstddef>
 #include <exception>
 #include <memory>
@@ -26,22 +27,18 @@
 
 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;
 
 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))
@@ -79,7 +76,7 @@ TransactionAcquire::done()
 }
 
 void
-TransactionAcquire::onTimer(bool progress, ScopedLockType& psl)
+TransactionAcquire::onTimer(bool progress, ScopedLockType&)
 {
     if (timeouts_ > kMaxTimeouts)
     {

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

@@ -11,6 +11,7 @@
 #include <xrpl/shamap/SHAMapAddNode.h>
 #include <xrpl/shamap/SHAMapTreeNode.h>
 
+#include <chrono>
 #include <cstddef>
 #include <memory>
 #include <utility>
@@ -20,14 +21,34 @@ 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 each timeout counted
+     * against the acquisition takes. Short, since a set is wanted for the
+     * consensus round that asked for it or not at all.
+     */
+    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. Defaulted in
+     *        production; TransactionAcquire_test passes a short one so a whole
+     *        timeout chain runs in a fraction of the time. TimeoutCounter
+     *        requires more than 10ms.
+     */
+    TransactionAcquire(
+        Application& app,
+        uint256 const& hash,
+        std::unique_ptr<PeerSet> peerSet,
+        std::chrono::milliseconds retryInterval = kRetryInterval);
     ~TransactionAcquire() override = default;
 
     SHAMapAddNode
@@ -41,14 +62,25 @@ public:
     void
     stillNeed();
 
-private:
+protected:
+    // Kept protected so a test subclass (see TransactionAcquire_test) can read the map's state,
+    // which nothing else publishes. Production callers reach a set through InboundTransactions.
     std::shared_ptr<SHAMap> map_;
+
+private:
     bool haveRoot_{false};
     std::unique_ptr<PeerSet> peerSet_;
 
     void
-    onTimer(bool progress, ScopedLockType& peerSetLock) override;
+    onTimer(bool progress, ScopedLockType& sl) override;
 
+    /**
+     * Settle the acquired set and hand it on, or report the failure. Call under
+     * mtx_.
+     *
+     * Runs at most once per outcome rather than once in total, since
+     * stillNeed() can revive a timed-out acquisition that then finishes.
+     */
     void
     done();