fix: Count partial batch progress in a TX-set reply

TransactionAcquire::takeNodes() accumulates one SHAMapAddNode across the
whole batch, so a packet that ends on a rejected node still counts the
nodes hooked in ahead of it, as InboundLedger::receiveNode() does. The
body moves to takeNodesLocked() and takeNodes() records progress on the
single exit, since several inner returns stop the batch early.

Progress is recorded for a batch that hooked a node in or answered with
one already held. A batch of nodes already held is what an honest second
responder to trigger()'s fan-out sends, so gotData() reads isGood()
rather than isUseful() when it decides whether to charge for useless
data.

include/xrpl/ledger/Ledger.h

@@ -105,6 +105,16 @@ public:
         std::vector<UInt256> const& amendments,
         Family& family);
 
+    /**
+     * Create a ledger from a header whose maps are filled in afterwards.
+     *
+     * Both maps start Synching against the hashes the header carries, which are
+     * input rather than derived, so setImmutable() leaves them alone.
+     *
+     * @param info The header to build from.
+     * @param rules The rules in force.
+     * @param family The SHAMap family the maps belong to.
+     */
     Ledger(LedgerHeader const& info, Rules rules, Family& family);
 
     /**
@@ -278,23 +288,25 @@ public:
     /**
      * Mark this ledger as immutable, so it can no longer be modified.
      *
-     * A ledger built or loaded locally cannot have an invalid map, since only
-     * a map syncing against hashes from outside can be proven impossible (see
-     * SHAMap::addKnownNode), so a caller on such a path may treat a false
-     * return as a broken internal invariant. A caller assembling a ledger from
-     * peer data may not: for it, false is an outcome a peer can produce.
+     * Only a map syncing against hashes from outside can be found unsound (see
+     * SHAMap::addKnownNode), so a caller whose ledger was built or loaded
+     * locally treats a false return as a broken internal invariant.
      *
      * @param rehash Whether to recompute the ledger hash from the header
      *        fields. The transaction and account hashes are recomputed from the
-     *        maps too, but only the first time.
+     *        maps too, but only the first time and only if the header did not
+     *        supply them.
      * @return false if either map is Invalid, leaving the immutable flag as it
-     *         was and the header untouched. A map invalidated partway through can
-     *         still leave the other one immutable, so a false return means the
-     *         ledger must be discarded rather than retried.
+     *         was and the header untouched. The ledger must then be discarded
+     *         rather than retried, since one map may already be immutable.
      */
     [[nodiscard]] bool
     setImmutable(bool rehash = true);
 
+    /**
+     * @return Whether both maps have been settled, so the ledger can no longer
+     *         change.
+     */
     bool
     isImmutable() const
     {
@@ -470,6 +482,15 @@ private:
 
     bool immutable_;
 
+    /**
+     * Whether the header supplied the transaction and account hashes, so they
+     * must not be derived from the maps.
+     *
+     * True only for a ledger built from a header, which stays verified against
+     * the hash it was asked for. Fixed at construction, unlike immutable_.
+     */
+    bool const mapHashesFromHeader_ = false;
+
     // A SHAMap containing the transactions associated with this ledger.
     SHAMap mutable txMap_;
 

src/libxrpl/ledger/Ledger.cpp

@@ -305,7 +305,8 @@ Ledger::Ledger(Ledger const& prevLedger, NetClock::time_point closeTime)
 }
 
 Ledger::Ledger(LedgerHeader const& info, Rules rules, Family& family)
-    : immutable_(true)
+    : immutable_(false)
+    , mapHashesFromHeader_(true)
     , txMap_(SHAMapType::TRANSACTION, info.txHash, family)
     , stateMap_(SHAMapType::STATE, info.accountHash, family)
     , rules_(std::move(rules))
@@ -337,31 +338,29 @@ Ledger::Ledger(
 bool
 Ledger::setImmutable(bool rehash)
 {
-    // A map found structurally invalid during sync must never be made immutable: isValid() tests
-    // only for Invalid, and an immutable ledger is treated as persistable. Asked before anything is
-    // written, so a refusal leaves the header exactly as it was rather than half relabeled.
+    // An immutable ledger is treated as persistable, so an invalid map must never become
+    // immutable. The validity test runs before anything is written, so a refusal leaves
+    // the header as it was.
     if (!mapsValid())
         return false;
 
-    // Read here but written to the header below, once the maps are settled: getHash() can unshare a
-    // dirty tree, so it has to run while the map is still mutable, while a write to the header must
-    // wait until both maps have made it. Skipped once the ledger is immutable, since its maps can
-    // no longer change.
-    bool const deriveMapHashes = !immutable_ && rehash;
+    // Read now but written to the header below: getHash() can unshare a dirty tree, so it must run
+    // while the map is still mutable, while the header may only be written once both maps are
+    // settled. Skipped when the ledger is already immutable or when the header supplied these.
+    bool const deriveMapHashes = !immutable_ && !mapHashesFromHeader_ && rehash;
     UInt256 const txHash = deriveMapHashes ? txMap_.getHash().asUInt256() : UInt256{};
     UInt256 const accountHash = deriveMapHashes ? stateMap_.getHash().asUInt256() : UInt256{};
 
-    // Both were valid at the check above, but a concurrent walk can invalidate one in between (see
-    // SHAMap::state_), so the result is checked. setInvalid() outranks Immutable and can land
-    // after both maps have been settled, so this narrows the window it guards.
+    // A concurrent walk can invalidate a map between the check above and here (see
+    // SHAMap::state_), so the result is checked. That narrows the window rather than closing it,
+    // since setInvalid() outranks Immutable.
     bool const bothImmutable = setMapsImmutable();
     SOMETIMES(!bothImmutable, "xrpl::Ledger::setImmutable : map invalidated while going immutable");
     if (!bothImmutable)
-        return false;
+        return false;  // LCOV_EXCL_LINE: only the walk named above reaches this, so no test does
 
-    // Written only now, so losing the race above leaves the header describing what the ledger was
-    // built from rather than a map that has since been abandoned. Forced rather than conditional,
-    // since this is the only place the hash transitions to valid.
+    // Written only now, so failing the check above leaves the header describing what the ledger was
+    // built from. Forced, since this is the only place the hash transitions to valid.
     if (deriveMapHashes)
     {
         header_.txHash = txHash;

src/test/app/InboundLedger_test.cpp

@@ -5,6 +5,7 @@
 #include <xrpld/app/ledger/InboundLedgers.h>
 #include <xrpld/app/ledger/LedgerMaster.h>
 
+#include <xrpl/basics/Blob.h>
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/basics/chrono.h>
 #include <xrpl/beast/unit_test/suite.h>
@@ -73,6 +74,37 @@ struct TestableInboundLedger final : InboundLedger
         trigger(nullptr, TriggerReason::Added);
     }
 
+    /**
+     * The same, as the timer chain does.
+     */
+    void
+    triggerTimeout()
+    {
+        trigger(nullptr, TriggerReason::Timeout);
+    }
+
+    /**
+     * Record how many timeouts have elapsed.
+     *
+     * @param timeouts The count to record.
+     */
+    void
+    setTimeouts(int timeouts)
+    {
+        ScopedLockType const sl(mtx_);
+        timeouts_ = timeouts;
+    }
+
+    /**
+     * Forget any recorded progress.
+     */
+    void
+    clearProgress()
+    {
+        ScopedLockType const sl(mtx_);
+        progress_ = false;
+    }
+
     /**
      * Record that every part has been fetched.
      */
@@ -360,6 +392,60 @@ struct InboundLedger_test : public beast::unit_test::Suite
         BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
     }
 
+    /**
+     * A ledger completed by a walk rather than by tryDB() is settled before it
+     * is reported complete.
+     *
+     * The order is visible only to a reader of isComplete(), which is read
+     * without mtx_, so this case pins the invariant instead: every ledger the
+     * acquisition reports is immutable.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testWalkSettlesBeforeReportingComplete(jtx::Env& env)
+    {
+        testcase("A ledger completed by a walk is settled before it is reported");
+
+        auto const chain = DeepChain::toLeaf(2, nextSeed());
+        auto const header = makeHeader(chain);
+
+        // Everything except the leaf, so tryDB() can root the state map but its walk still has
+        // something to ask for, which leaves the completion to trigger().
+        storeHeader(env, header);
+        storeStateNodes(env, header, chain, chain.deepestDepth - 1);
+
+        auto const acquire = std::make_shared<TestableInboundLedger>(
+            env.app(),
+            header.hash,
+            header.seq,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::make_unique<RequestCountingPeerSet>());
+
+        BEAST_EXPECT(!acquire->checkLocal());
+        BEAST_EXPECT(!acquire->isComplete());
+        BEAST_EXPECT(!acquire->isFailed());
+
+        // The walk can finish only now, which places the completion in the walk.
+        storeStateNodes(env, header, chain, chain.deepestDepth);
+
+        acquire->triggerAdded();
+
+        BEAST_EXPECT(acquire->isComplete());
+        BEAST_EXPECT(!acquire->isFailed());
+
+        auto const settled = acquire->getLedger();
+        BEAST_EXPECT(settled != nullptr);
+        if (settled)
+            BEAST_EXPECT(settled->isImmutable());
+
+        // done()'s success arm ran, so the ledger reached LedgerMaster and the hash is absent
+        // from the failure list.
+        BEAST_EXPECT(env.app().getLedgerMaster().getLedgerByHash(header.hash) != nullptr);
+        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
+    }
+
     /**
      * A ledger whose map goes invalid on the way to being settled must be
      * discarded rather than delivered.
@@ -502,6 +588,249 @@ struct InboundLedger_test : public beast::unit_test::Suite
             waitFor([&] { return env.app().getInboundLedgers().isFailure(otherHeader.hash); }));
     }
 
+    /**
+     * A ledger assembled from local data must be judged even when only
+     * one map is settled.
+     *
+     * tryDB() walks both maps to see what is on hand, and a fetch pack is
+     * checked against each node's own hash rather than the shape it
+     * implies, so a whole chain can resolve locally without passing
+     * through addKnownNode().
+     *
+     * The asymmetry is the point: the transaction map is the chain, so
+     * its walk abandons it, while the state root is a hash no fetch pack
+     * supplies, leaving that map merely incomplete. tryDB() therefore
+     * sets neither flag and has to reach the verdict itself, since the
+     * setImmutable() call further down needs both.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testLocalChainFailsAcquire(jtx::Env& env)
+    {
+        testcase("A chain found locally fails the acquire");
+
+        DeepChain const chain{nextSeed()};
+
+        // The chain as the transaction root, and an arbitrary hash, seeded nowhere, as the state
+        // root.
+        auto const header = makeHeader(chain.rootHash.asUInt256(), UInt256{99});
+        auto& ledgerMaster = env.app().getLedgerMaster();
+
+        // The header, prefixed the way tryDB() expects to find it in a fetch pack.
+        Serializer hs;
+        hs.add32(HashPrefix::LedgerMaster);
+        addRaw(header, hs);
+        ledgerMaster.addFetchPack(header.hash, std::make_shared<Blob>(hs.modData()));
+
+        // Every node of the chain, keyed by its own hash. TransactionStateSF::getNode() reads
+        // these, so the transaction-map walk resolves the whole chain with no peer involved.
+        for (auto depth = 0u; depth <= SHAMap::kLeafDepth; ++depth)
+        {
+            ledgerMaster.addFetchPack(
+                chain.nodeAt(depth)->getHash().asUInt256(),
+                std::make_shared<Blob>(chain.prefixedNodeAt(depth)));
+        }
+
+        auto const acquire = std::make_shared<InboundLedger>(
+            env.app(),
+            header.hash,
+            header.seq,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::make_unique<RequestCountingPeerSet>());
+
+        // checkLocal() routes into tryDB() with no peer data. It reports true only because the
+        // acquisition ended.
+        BEAST_EXPECT(acquire->checkLocal());
+
+        BEAST_EXPECT(acquire->isFailed());
+        BEAST_EXPECT(!acquire->isComplete());
+
+        // A failed acquisition reports no ledger, though it still holds the partial one it built.
+        BEAST_EXPECT(acquire->getLedger() == nullptr);
+    }
+
+    /**
+     * The aggressive-retry branch of trigger() must judge a map the walk
+     * abandoned.
+     *
+     * That branch reads an empty getNeededHashes() result as a complete map,
+     * and the walk it runs can reach the invalid verdict itself once nodes
+     * resolve from local storage. Only the root is local when tryDB() runs, so
+     * the state map holds a root and the walk stops one level down. The branch
+     * also needs a timeout count above kLedgerBecomeAggressiveThreshold, which
+     * the case records directly rather than waiting out the timer chain.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testAggressiveRetryJudgesLocalMap(jtx::Env& env)
+    {
+        testcase("An aggressive retry judges a map the walk abandoned");
+
+        DeepChain const chain{nextSeed()};
+
+        // The chain as the state root, and no transactions, so only the state map is in play.
+        auto const header = makeHeader(chain);
+        auto& ledgerMaster = env.app().getLedgerMaster();
+
+        Serializer hs;
+        hs.add32(HashPrefix::LedgerMaster);
+        addRaw(header, hs);
+        ledgerMaster.addFetchPack(header.hash, std::make_shared<Blob>(hs.modData()));
+
+        // Only the root, so the state map gets a root but the walk stops one level down.
+        ledgerMaster.addFetchPack(
+            chain.nodeAt(0)->getHash().asUInt256(),
+            std::make_shared<Blob>(chain.prefixedNodeAt(0)));
+
+        auto const acquire = std::make_shared<TestableInboundLedger>(
+            env.app(),
+            header.hash,
+            header.seq,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::make_unique<RequestCountingPeerSet>());
+
+        // The acquisition is alive: it has the header and a state root, and still wants the rest.
+        BEAST_EXPECT(!acquire->checkLocal());
+        BEAST_EXPECT(!acquire->isFailed());
+        BEAST_EXPECT(acquire->getJson(0)[jss::have_header].asBool());
+        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());
+
+        auto const ledger = mutableLedger(*acquire);
+        BEAST_EXPECT(ledger != nullptr);
+        if (!ledger)
+            return;
+        BEAST_EXPECT(ledger->stateMap().isValid());
+
+        // The rest of the chain becomes resolvable only now, which places the verdict in this walk.
+        for (auto depth = 1u; depth <= SHAMap::kLeafDepth; ++depth)
+        {
+            ledgerMaster.addFetchPack(
+                chain.nodeAt(depth)->getHash().asUInt256(),
+                std::make_shared<Blob>(chain.prefixedNodeAt(depth)));
+        }
+
+        // kLedgerBecomeAggressiveThreshold is 4 and file-local, so name the requirement here.
+        acquire->setTimeouts(5);
+        acquire->clearProgress();
+        acquire->triggerTimeout();
+
+        // The walk resolved the chain locally and abandoned the map, and trigger() recorded that
+        // rather than reading the empty result as a finished acquisition.
+        BEAST_EXPECT(!ledger->stateMap().isValid());
+        BEAST_EXPECT(acquire->isFailed());
+        BEAST_EXPECT(!acquire->isComplete());
+
+        // have_state is the discriminating assertion: only this guard leaves it false, since every
+        // have-flag is set on the way to the setImmutable() backstop in done().
+        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());
+
+        // The same branch with no header, which is the other arm of hasInvalidMap(): no map, so
+        // the arm answers false. getNeededHashes() has asked for the header, so the non-empty
+        // branch is the right one and the acquisition stays alive.
+        auto const headerless = std::make_shared<TestableInboundLedger>(
+            env.app(),
+            UInt256{7},
+            0,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::make_unique<RequestCountingPeerSet>());
+
+        headerless->setTimeouts(5);
+        headerless->clearProgress();
+        headerless->triggerTimeout();
+
+        BEAST_EXPECT(mutableLedger(*headerless) == nullptr);
+        BEAST_EXPECT(!headerless->isFailed());
+        BEAST_EXPECT(!headerless->isComplete());
+    }
+
+    /**
+     * The ordinary trigger() path must judge a map its own walk abandoned.
+     *
+     * Covers the state-map walk trigger() runs with mtx_ released, where the
+     * case above covers the empty getNeededHashes() branch. That verdict has to
+     * be read before the flags below it. Staged as that case is, with the
+     * timeout count left at zero to stay off the aggressive branch.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testWalkJudgesMapOnOrdinaryTrigger(jtx::Env& env)
+    {
+        testcase("An ordinary trigger judges a map its walk abandoned");
+
+        DeepChain const chain{nextSeed()};
+
+        auto const header = makeHeader(chain);
+        auto& ledgerMaster = env.app().getLedgerMaster();
+
+        Serializer hs;
+        hs.add32(HashPrefix::LedgerMaster);
+        addRaw(header, hs);
+        ledgerMaster.addFetchPack(header.hash, std::make_shared<Blob>(hs.modData()));
+
+        // Only the root, so the state map gets a root but the walk stops one level down.
+        ledgerMaster.addFetchPack(
+            chain.nodeAt(0)->getHash().asUInt256(),
+            std::make_shared<Blob>(chain.prefixedNodeAt(0)));
+
+        auto peerSet = std::make_unique<RequestCountingPeerSet>();
+        auto const* const peerSetPtr = peerSet.get();
+
+        auto const acquire = std::make_shared<TestableInboundLedger>(
+            env.app(),
+            header.hash,
+            header.seq,
+            InboundLedger::Reason::GENERIC,
+            stopwatch(),
+            std::move(peerSet));
+
+        BEAST_EXPECT(!acquire->checkLocal());
+        BEAST_EXPECT(!acquire->isFailed());
+        BEAST_EXPECT(acquire->getJson(0)[jss::have_header].asBool());
+        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());
+
+        auto const ledger = mutableLedger(*acquire);
+        BEAST_EXPECT(ledger != nullptr);
+        if (!ledger)
+            return;
+        BEAST_EXPECT(ledger->stateMap().isValid());
+
+        // The rest of the chain becomes resolvable only now, which places the verdict in this walk.
+        for (auto depth = 1u; depth <= SHAMap::kLeafDepth; ++depth)
+        {
+            ledgerMaster.addFetchPack(
+                chain.nodeAt(depth)->getHash().asUInt256(),
+                std::make_shared<Blob>(chain.prefixedNodeAt(depth)));
+        }
+
+        // Sound going in, which is the first half of the tripwire below.
+        BEAST_EXPECT(ledger->stateMap().isValid());
+
+        int const requestsBefore = peerSetPtr->requests();
+        acquire->triggerAdded();
+
+        // The walk resolved the chain locally and abandoned the map, and trigger() recorded that
+        // rather than reading the empty node list as a finished state map.
+        BEAST_EXPECT(!ledger->stateMap().isValid());
+        BEAST_EXPECT(acquire->isFailed());
+
+        // have_state is the discriminating assertion. isComplete() is not: it reads false whether
+        // this guard fired or the setImmutable() backstop in done() did.
+        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());
+
+        // A tripwire that keeps this case on the branch it names: the aggressive branch needs both
+        // a TIMEOUT reason and a count above kLedgerBecomeAggressiveThreshold.
+        BEAST_EXPECT(acquire->getJson(0)[jss::timeouts].asInt() == 0);
+
+        // The request count is unchanged after the verdict, so the guard ended the round.
+        BEAST_EXPECT(peerSetPtr->requests() == requestsBefore);
+    }
+
     /**
      * A peer-supplied header whose account hash is zero fails the acquisition
      * and signals, as a local one does. The peer is not charged, and
@@ -604,17 +933,10 @@ struct InboundLedger_test : public beast::unit_test::Suite
     /**
      * 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.
-     *
-     * The store is empty for this hash and no reply arrives, so every tick
-     * counts a timeout and the count climbs to the limit. A hash of its own,
-     * so this case owns its entry in the failure list.
+     * The only case that drives onTimer() rather than trigger() directly. Past
+     * kLedgerTimeoutRetriesMax the acquisition fails itself and done() records
+     * that, so the same ledger is not asked for again. The store is empty for
+     * this hash and no reply arrives, so every tick counts a timeout.
      *
      * @param env The environment to run in.
      */
@@ -662,7 +984,11 @@ struct InboundLedger_test : public beast::unit_test::Suite
         BEAST_EXPECT(!acquire->isComplete());
         BEAST_EXPECT(peerSetPtr->requests() > requestsFromInit);
 
-        // done() remembered the hash, which is what stops the next round asking again.
+        // done() remembered the hash. This case reaches getLedger() with no ledger ever built, so
+        // both arms of the failed_ gate answer null here. The local-chain case above is what
+        // covers the gate itself: tryDB() builds a partial ledger and then fails, so the gate is
+        // the only reason the answer is null.
+        BEAST_EXPECT(acquire->getLedger() == nullptr);
         BEAST_EXPECT(
             waitFor([&] { return env.app().getInboundLedgers().isFailure(kUnknownLedger); }));
     }
@@ -676,10 +1002,14 @@ struct InboundLedger_test : public beast::unit_test::Suite
         jtx::Env env{*this};
 
         testLocalLedgerCompletesAcquire(env);
+        testWalkSettlesBeforeReportingComplete(env);
         testInvalidatedLedgerFailsInDone(env);
         testLocalFailureSignalsDone(env);
         testPeerZeroAccountHashFails(env);
         testPeerHeaderWithoutTransactionsCompletes(env);
+        testLocalChainFailsAcquire(env);
+        testAggressiveRetryJudgesLocalMap(env);
+        testWalkJudgesMapOnOrdinaryTrigger(env);
 
         // Last: the only case that waits out a whole timeout chain.
         testTimerRetriesThenGivesUp(env);

src/test/app/TransactionAcquire_test.cpp

@@ -11,12 +11,14 @@
 #include <xrpl/shamap/SHAMap.h>
 #include <xrpl/shamap/SHAMapAddNode.h>
 #include <xrpl/shamap/SHAMapNodeID.h>
+#include <xrpl/shamap/SHAMapTreeNode.h>
 
 #include <xrpl.pb.h>
 
 #include <chrono>
 #include <memory>
 #include <set>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -403,6 +405,79 @@ struct TransactionAcquire_test : public beast::unit_test::Suite
         BEAST_EXPECT(peer->charges() == std::vector{resource::kFeeInvalidData});
     }
 
+    /**
+     * A batch that ends on a bad node still counts the good nodes ahead of
+     * it, and a batch of nodes already held still counts as an answer.
+     *
+     * The recorded progress is what the verdict is for: it stops the
+     * next timer tick from counting a timeout, so a batch that hooked
+     * any node reports progress. A batch of nodes already held counts as
+     * well, because the peer answered what it was asked, and that answer
+     * earns the same postponement.
+     * The flag is read rather than the returned tally, which only stands
+     * in for it, and cleared between batches so each reading is about the
+     * batch just fed.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testPartialBatchIsCounted(jtx::Env& env)
+    {
+        testcase("A batch ending on a bad node still counts the good nodes");
+
+        DeepChain const chain{nextSeed()};
+
+        auto const acquire = std::make_shared<TestableTransactionAcquire>(
+            env.app(), chain.rootHash.asUInt256(), std::make_unique<RequestCountingPeerSet>());
+        auto const peer = std::make_shared<ChargeRecordingPeer>();
+
+        // The root is useful, so it records progress.
+        static_cast<void>(acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer));
+        BEAST_EXPECT(acquire->madeProgress());
+        acquire->clearProgress();
+
+        // Good nodes at depths 1 and 2, then a further chain node mislabeled at a position only a
+        // leaf may occupy. The map stays sound, so only the last node is bad.
+        std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> batch;
+        batch.emplace_back(chain.idAt(1), chain.nodeAt(1));
+        batch.emplace_back(chain.idAt(2), chain.nodeAt(2));
+        batch.emplace_back(SHAMapNodeID{9, UInt256{}}, chain.nodeAt(3));
+
+        auto const san = acquire->takeNodes(batch, peer);
+
+        // The verdict names both halves. The batch stops on the bad node, so one bad node is
+        // counted however many were left unexamined behind it.
+        BEAST_EXPECTS(tallyIs(san, 2, 1, 0), san.get());
+        BEAST_EXPECT(san.isUseful());
+        BEAST_EXPECT(acquire->madeProgress());
+        BEAST_EXPECT(acquire->isMapValid());
+
+        // A batch holding only the root we already have: counted as a duplicate. The duplicate
+        // tally is what makes it an answer, so it postpones the timeout.
+        acquire->clearProgress();
+        auto const repeatedRoot = acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer);
+
+        BEAST_EXPECTS(tallyIs(repeatedRoot, 0, 0, 1), repeatedRoot.get());
+        BEAST_EXPECT(repeatedRoot.isGood());
+        BEAST_EXPECT(!repeatedRoot.isUseful());
+        BEAST_EXPECT(acquire->madeProgress());
+
+        // The same root alongside a node we do need: the duplicate is reported as one, and the
+        // node that did hook in is what records the progress.
+        std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> mixed;
+        mixed.emplace_back(SHAMapNodeID{}, chain.nodeAt(0));
+        mixed.emplace_back(SHAMapNodeID{3, UInt256{}}, chain.nodeAt(3));
+
+        auto const withDuplicateRoot = acquire->takeNodes(mixed, peer);
+
+        BEAST_EXPECTS(tallyIs(withDuplicateRoot, 1, 0, 1), withDuplicateRoot.get());
+        BEAST_EXPECT(withDuplicateRoot.isUseful());
+        BEAST_EXPECT(acquire->madeProgress());
+
+        // takeNodes() charges nobody: InboundTransactions::gotData() reads the verdict and decides.
+        BEAST_EXPECT(peer->charges().empty());
+    }
+
     /**
      * init() passes hasTxSet() to addPeers() as its candidate filter.
      *
@@ -445,14 +520,119 @@ struct TransactionAcquire_test : public beast::unit_test::Suite
         BEAST_EXPECT(peerSetPtr->requests() >= 1);
     }
 
+    /**
+     * A timed-out acquisition asks again, and examines data again, once
+     * stillNeed() revives it.
+     *
+     * The pending timer is private to TimeoutCounter, so the case lets a real
+     * timeout chain run rather than calling cancel().
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testRevivedAcquireCanRequestAgain(jtx::Env& env)
+    {
+        testcase("A revived acquire asks again and accepts data again");
+
+        DeepChain const chain{nextSeed()};
+
+        // One candidate, offered once by init(). RequestCountingPeerSet dedups by tracked id like
+        // the real peer set, so a further request to it comes from onTimer()'s broadcast
+        // trigger(nullptr).
+        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();
+
+        // A short interval, since this case waits out a whole timeout chain.
+        auto const acquire = std::make_shared<TransactionAcquire>(
+            env.app(), chain.rootHash.asUInt256(), std::move(peerSet), kFastRetry);
+
+        acquire->init(1);
+        BEAST_EXPECT(waitFor([&] { return peerSetPtr->requests() > 0; }));
+        BEAST_EXPECT(peerSetPtr->addedPeers() == std::set<Peer::ID>{candidate->id()});
+
+        // A root from a different chain, so polling with it leaves haveRoot_ false and the
+        // acquisition fails on its own. See testTimerBroadcastsThenGivesUp, which probes the same
+        // way.
+        DeepChain const wrongChain{nextSeed()};
+        auto const probe = [&] {
+            return wasIgnored(acquire->takeNodes(
+                {{SHAMapNodeID{}, wrongChain.nodeAt(0)}}, std::make_shared<ChargeRecordingPeer>()));
+        };
+
+        // Progress stays clear, so onTimer() counts a timeout every tick; past kMaxTimeouts (20)
+        // it fails itself and stops examining data.
+        BEAST_EXPECT(waitFor(probe));
+        int const requestsBeforeRevival = peerSetPtr->requests();
+
+        // Revived, so the timer chain restarts. stillNeed() clamps timeouts_ down to
+        // kNormTimeouts rather than to zero, so the very next tick broadcasts to every peer
+        // already tracked - the only way a peer already selected once is asked again, and so
+        // what shows the timer chain was restarted rather than just the failed flag cleared.
+        acquire->stillNeed();
+        BEAST_EXPECT(waitFor([&] { return peerSetPtr->requests() > requestsBeforeRevival; }));
+
+        // Data is examined again too: the real root is accepted, and asks for the next level.
+        auto const peer = std::make_shared<ChargeRecordingPeer>();
+        auto const revived = acquire->takeNodes({{SHAMapNodeID{}, chain.nodeAt(0)}}, peer);
+        BEAST_EXPECT(!wasIgnored(revived));
+        BEAST_EXPECT(revived.isUseful());
+
+        // Stop the retry loop, which asks for as long as the acquisition runs.
+        acquire->cancel();
+    }
+
+    /**
+     * A running acquisition keeps the wait it already has.
+     *
+     * setTimer() cancels any pending wait, so calling stillNeed() faster than
+     * the interval is what makes the early return observable. Keeps the
+     * production interval, so the asking is clearly faster than the wait.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testStillNeedLeavesARunningAcquireAlone(jtx::Env& env)
+    {
+        testcase("A running acquire keeps the wait it has");
+
+        // One candidate, so every tick that survives produces a request.
+        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();
+
+        // An unrelated hash, so the acquisition stays incomplete and keeps asking.
+        auto const acquire =
+            std::make_shared<TransactionAcquire>(env.app(), UInt256{43}, std::move(peerSet));
+
+        // init() asks the candidate once and arms the timer. That first request is not the one
+        // under test, so count from here.
+        acquire->init(1);
+        int const requestsFromInit = peerSetPtr->requests();
+
+        // Ask again far faster than the interval, as a short consensus round does. Every ask clamps
+        // the timeout count, so the acquisition keeps running throughout.
+        auto const askAgainRepeatedly = [&] {
+            acquire->stillNeed();
+            std::this_thread::sleep_for(std::chrono::milliseconds{20});
+            return peerSetPtr->requests() > requestsFromInit;
+        };
+
+        // A tick gets through despite the asking, which is what the kept wait allows.
+        BEAST_EXPECT(waitFor(askAgainRepeatedly));
+
+        acquire->cancel();
+    }
+
     /**
      * The retry timer broadcasts 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.
-     * The broadcast count is read after the give-up, which the failed_ guard
-     * in trigger() makes safe.
+     * Pins the two behaviors rather than the thresholds they trip at, since
+     * bounding those means asserting on wall clock. The broadcast count is
+     * read after the give-up, which the failed_ guard in trigger() makes safe.
      *
      * The acquisition tracks no peer, so the broadcast this case counts reaches
      * nobody in production either. What is pinned is that onTimer() issues it.
@@ -510,9 +690,13 @@ struct TransactionAcquire_test : public beast::unit_test::Suite
         testDuplicateNonRootReplyIsFree(env);
         testUndeserializableNodeIsCharged(env);
         testMismatchedRootThroughGotDataIsCharged(env);
+        testPartialBatchIsCounted(env);
         testInitFiltersCandidatesByHasTxSet(env);
+        testStillNeedLeavesARunningAcquireAlone(env);
 
-        // Last: the only case that waits out a whole timeout chain.
+        // Last: both wait out a whole timeout chain, and neither needs the cases above to have
+        // run first.
+        testRevivedAcquireCanRequestAgain(env);
         testTimerBroadcastsThenGivesUp(env);
     }
 

src/tests/libxrpl/shamap/SHAMapSync.cpp

@@ -536,9 +536,63 @@ TEST_F(SHAMapSyncTest, refused_settle_leaves_the_header_alone)
     EXPECT_EQ(ledger.header().hash, hashBefore);
 }
 
-// Invalid is terminal: setImmutable() and clearSynching() offer no way back out of it, however
-// many times they are called. setSynching() is left alone, since it is unreachable on an invalid
-// map today and says so with an UNREACHABLE.
+// A ledger built from a header must not claim to be immutable before setImmutable() has found both
+// maps sound: they start out Synching and are filled in afterwards, and LedgerHistory::insert() and
+// LedgerReplayMsgHandler both gate on that claim to catch exactly that case.
+TEST_F(SHAMapSyncTest, ledger_from_header_is_not_immutable_until_settled)
+{
+    TestNodeFamily f{j_};
+    DeepChain const chain;
+
+    LedgerHeader header;
+    header.seq = 2;
+    header.txHash = chain.rootHash.asUInt256();
+    header.hash = calculateLedgerHash(header);
+
+    Ledger ledger{header, noAmendments(), f};
+
+    // Both maps are still syncing, so the ledger reports mutable.
+    EXPECT_TRUE(ledger.txMap().isSynching());
+    EXPECT_TRUE(ledger.stateMap().isSynching());
+    EXPECT_FALSE(ledger.isImmutable());
+
+    ASSERT_TRUE(chain.fill(ledger.txMap()));
+    ASSERT_TRUE(chain.addOffendingNode(ledger.txMap()).isInvalid());
+    ASSERT_FALSE(ledger.txMap().isValid());
+
+    EXPECT_FALSE(ledger.setImmutable());
+    EXPECT_FALSE(ledger.isImmutable());
+}
+
+// The header's own map hashes are what the maps are synced against, so settling keeps them. The
+// transaction map is left empty while the header names a chain root, so the two differ and the
+// kept value is observable.
+TEST_F(SHAMapSyncTest, ledger_from_header_keeps_the_map_hashes_it_was_given)
+{
+    TestNodeFamily f{j_};
+    DeepChain const chain;
+
+    LedgerHeader header;
+    header.seq = 2;
+    header.txHash = chain.rootHash.asUInt256();
+    header.hash = calculateLedgerHash(header);
+    auto const verifiedHash = header.hash;
+
+    Ledger ledger{header, noAmendments(), f};
+    ASSERT_FALSE(ledger.isImmutable());
+
+    // The map is empty, so it hashes to zero. Both maps are still valid, so settling succeeds.
+    ASSERT_TRUE(ledger.txMap().getHash().isZero());
+    ASSERT_TRUE(ledger.setImmutable());
+    EXPECT_TRUE(ledger.isImmutable());
+
+    EXPECT_EQ(ledger.header().txHash, chain.rootHash.asUInt256());
+    EXPECT_EQ(ledger.header().hash, verifiedHash);
+}
+
+// Invalid is terminal: neither setImmutable() nor clearSynching() leads back out of it, however
+// many times they are called. setSynching() is left out, since an invalid map cannot reach it,
+// which its UNREACHABLE states.
 TEST_F(SHAMapSyncTest, invalid_state_is_terminal)
 {
     TestNodeFamily f{j_};

src/xrpld/app/ledger/InboundLedger.h

@@ -75,7 +75,11 @@ public:
     update(std::uint32_t seq);
 
     /**
-     * Returns true if we got all the data.
+     * Whether the acquisition succeeded and its ledger has been settled. Every
+     * path that sets this settles the ledger first, so a caller that sees it
+     * may use the ledger directly.
+     *
+     * @return Whether the ledger is complete and settled.
      */
     bool
     isComplete() const
@@ -84,7 +88,7 @@ public:
     }
 
     /**
-     * Returns false if we failed to get the data.
+     * @return Whether the acquisition has failed.
      */
     bool
     isFailed() const
@@ -161,8 +165,10 @@ protected:
     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.
+     * Settle the acquisition, publish its outcome, and signal whatever is
+     * waiting on it. Runs at most once. Call under mtx_, which the flags
+     * written here require. Settles before publishing, since isComplete() is
+     * read without mtx_.
      */
     void
     done();
@@ -180,6 +186,20 @@ private:
     void
     tryDB(node_store::Database& srcDB);
 
+    /**
+     * Whether either map of the ledger being acquired has been found invalid.
+     *
+     * A walk returns a bare list of hashes, so this is what tells a satisfied
+     * map from an abandoned one. Callers ask it before reading emptiness as
+     * "nothing left to fetch". See SHAMap::addKnownNode for why the verdict is
+     * final.
+     *
+     * @return Whether either map is Invalid, and false while there is no ledger
+     *         yet, since then there is no map to judge.
+     */
+    [[nodiscard]] bool
+    hasInvalidMap() const;
+
     void
     onTimer(bool progress, ScopedLockType& sl) override;
 

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

@@ -113,8 +113,8 @@ InboundLedger::init(ScopedLockType& collectionLock)
     XRPL_ASSERT(
         ledger_->header().seq < kXrpLedgerEarliestFees || FeeSettingsEntryR(*ledger_),
         "xrpl::InboundLedger::init : valid ledger fees");
-    // tryDB() verified both maps before setting complete_ and mtx_ has been held since, so
-    // nothing can have invalidated them.
+    // tryDB() verified both maps before setting complete_ and mtx_ has been held since, so both
+    // maps are still sound here.
     if (!ledger_->setImmutable())
     {
         // LCOV_EXCL_START
@@ -237,6 +237,12 @@ InboundLedger::neededStateHashes(int max, SHAMapSyncFilter const* filter) const
     return neededHashes(ledger_->header().accountHash, ledger_->stateMap(), max, filter);
 }
 
+bool
+InboundLedger::hasInvalidMap() const
+{
+    return ledger_ && !ledger_->mapsValid();
+}
+
 // See how much of the ledger data is stored locally
 // Data found in a fetch pack will be stored
 void
@@ -340,13 +346,23 @@ InboundLedger::tryDB(node_store::Database& srcDB)
         }
     }
 
+    // Judged here rather than at the setImmutable() below, which runs only once both flags are
+    // set: one map can be abandoned while the other is merely incomplete.
+    if (hasInvalidMap())
+    {
+        JLOG(journal_.warn()) << "Ledger " << hash_ << " found locally has an invalid map";
+        failed_ = true;
+        return;
+    }
+
     if (haveTransactions_ && haveState_)
     {
         XRPL_ASSERT(
             ledger_->header().seq < kXrpLedgerEarliestFees || FeeSettingsEntryR(*ledger_),
             "xrpl::InboundLedger::tryDB : valid ledger fees");
-        // Settled before complete_ is published, so a caller that reads the flag never sees a
-        // ledger this function has not finished with.
+        // Settled before complete_ is published, so a caller that reads the flag sees a ledger
+        // this function has finished with. Reached despite the guard above because trigger()
+        // walks the state map with mtx_ released, so that walk can reach the verdict in between.
         if (!ledger_->setImmutable())
         {
             JLOG(journal_.warn()) << "Ledger " << hash_ << " found locally is invalid";
@@ -441,28 +457,34 @@ InboundLedger::done()
     signaled_ = true;
     touch();
 
-    // Settled before the outcome below is logged or acted on, so nothing reports a ledger this
-    // function has since refused.
-    if (complete_ && !failed_ && ledger_)
+    // Settled here, and complete_ published only once it is settled. isComplete() is read without
+    // mtx_, by InboundLedgers::acquire() among others, and LedgerHistory::insert() and
+    // LedgerHolder::set() each require an immutable ledger. tryDB() already settles its own
+    // result and sets complete_ itself, so that path arrives here with the ledger immutable and
+    // only the reporting below left to do.
+    bool const haveEverything = haveHeader_ && haveState_ && haveTransactions_;
+    if (!failed_ && ledger_ && (complete_ || haveEverything))
     {
         XRPL_ASSERT(
             ledger_->header().seq < kXrpLedgerEarliestFees || FeeSettingsEntryR(*ledger_),
             "xrpl::InboundLedger::done : valid ledger fees");
-        // Recovers rather than asserting: peer data produces this verdict, so a caller cannot know
-        // its map is still sound. Recovering rather than asserting is therefore the contract.
-        // Best-effort even so: setInvalid() outranks Immutable, so a walk that reaches the verdict
-        // after both maps have been settled leaves an immutable ledger with an invalid map. It
-        // narrows the window rather than closing it.
+        // trigger() walks the state map with mtx_ released, so that walk can reach the verdict
+        // after the flags said there was nothing left to fetch. A race rather than a broken
+        // invariant, and one that peer data produces, so this recovers rather than asserts.
+        // setInvalid() outranks Immutable, so a walk that reaches the verdict after both maps
+        // have been settled leaves an immutable ledger with an invalid map.
+        SOMETIMES(hasInvalidMap(), "xrpl::InboundLedger::done : map invalidated by a race");
         if (!ledger_->setImmutable())
         {
             JLOG(journal_.warn()) << "Acquired ledger " << hash_ << " is invalid";
-            // Withdrawn as well as failed, so a caller that already read complete_ - or that checks
-            // it before failed_ - cannot go on treating this ledger as delivered.
+            // Withdrawn as well as failed, so a caller that already read complete_, or that
+            // checks it before failed_, sees the claim withdrawn.
             complete_ = false;
             failed_ = true;
         }
         else
         {
+            complete_ = true;
             switch (reason_)
             {
                 case Reason::HISTORY:
@@ -485,9 +507,11 @@ InboundLedger::done()
 
     // mtx_ is held, so this may only post the work rather than do it.
     app_.getJobQueue().addJob(JtLedgerData, "AcqDone", [self = shared_from_this()] {
-        if (self->complete_ && !self->failed_)
+        // Read through getLedger(), which consults failed_ itself, so failure and a null ledger_
+        // are both refused by one check. checkAccept() requires a non-null ledger.
+        if (auto const ledger = self->getLedger(); self->complete_ && ledger)
         {
-            self->app_.getLedgerMaster().checkAccept(self->getLedger());
+            self->app_.getLedgerMaster().checkAccept(ledger);
             self->app_.getLedgerMaster().tryAdvance();
         }
         else
@@ -553,7 +577,20 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         {
             auto const need = getNeededHashes();
 
-            if (!need.empty())
+            // The validity check runs ahead of the emptiness test, since getNeededHashes() walks
+            // both maps and can reach the verdict itself. The result is read once, so the hint
+            // below and the test it feeds share one observation of a map another thread can be
+            // invalidating. The claim is withdrawn alongside the failure, so failed_ and
+            // complete_ never both read true.
+            bool const invalidMap = hasInvalidMap();
+            SOMETIMES(invalidMap, "xrpl::InboundLedger::trigger : map is invalid");
+            if (invalidMap)
+            {
+                JLOG(journal_.warn()) << "Acquire " << hash_ << " has an invalid map";
+                failed_ = true;
+                complete_ = false;
+            }
+            else if (!need.empty())
             {
                 protocol::TMGetObjectByHash tmBH;
                 bool typeSet = false;
@@ -590,11 +627,11 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
             }
             else
             {
+                // The tail of this function settles the ledger and reports it complete.
                 JLOG(journal_.info()) << "getNeededHashes says acquire is complete";
                 haveHeader_ = true;
                 haveTransactions_ = true;
                 haveState_ = true;
-                complete_ = true;
             }
         }
     }
@@ -657,27 +694,31 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         {
             AccountStateSF filter(ledger_->stateMap().family().db(), app_.getLedgerMaster());
 
-            // Release the lock while we process the large state map
+            // Release the lock while the state map is walked. mtx_ is recursive, so an sl.unlock()
+            // under onTimer() or the addPeers() callback leaves mtx_ held. The flags are re-read
+            // below because another packet can be handled while this walk runs.
             sl.unlock();
             auto nodes = ledger_->stateMap().getMissingNodes(kMissingNodesFind, &filter);
             sl.lock();
 
+            // The validity check runs outside the flags' guard below, since the verdict is about
+            // the map rather than this round: it holds even if another thread reported the ledger
+            // complete while the lock was released.
+            bool const walkAbandonedMap = hasInvalidMap();
+            SOMETIMES(walkAbandonedMap, "xrpl::InboundLedger::trigger : map abandoned by its walk");
+            if (walkAbandonedMap)
+            {
+                JLOG(journal_.warn()) << "Ledger " << hash_ << " has a map its walk abandoned";
+                failed_ = true;
+                complete_ = false;
+            }
             // Make sure nothing happened while we released the lock
-            if (!failed_ && !complete_ && !haveState_)
+            else if (!failed_ && !complete_ && !haveState_)
             {
                 if (nodes.empty())
                 {
-                    if (!ledger_->stateMap().isValid())
-                    {
-                        failed_ = true;
-                    }
-                    else
-                    {
-                        haveState_ = true;
-
-                        if (haveTransactions_)
-                            complete_ = true;
-                    }
+                    // The test above already caught a map the walk abandoned, so this one is sound.
+                    haveState_ = true;
                 }
                 else
                 {
@@ -739,9 +780,6 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
                 else
                 {
                     haveTransactions_ = true;
-
-                    if (haveState_)
-                        complete_ = true;
                 }
             }
             else
@@ -766,12 +804,14 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
         }
     }
 
-    if (complete_ || failed_)
+    // Having every part is not yet a completed acquisition: done() settles the ledger first and
+    // only then publishes complete_. Called with mtx_ still held, as done() documents, so the flags
+    // it writes are not written unlocked; mtx_ is recursive, so a caller that already holds it is
+    // unaffected.
+    if (failed_ || (haveHeader_ && haveState_ && haveTransactions_))
     {
-        JLOG(journal_.debug()) << "Done:" << (complete_ ? " complete" : "")
-                               << (failed_ ? " failed " : " ") << ledger_->header().seq;
-        // Called with mtx_ still held, so the flags done() writes are not written unlocked; mtx_
-        // is recursive, so a caller that already holds it further up is unaffected.
+        JLOG(journal_.debug()) << "Done:" << (failed_ ? " failed " : " have everything ")
+                               << ledger_->header().seq;
         done();
     }
 }
@@ -988,11 +1028,10 @@ InboundLedger::receiveNode(
             haveState_ = true;
         }
 
+        // done() settles the ledger before publishing complete_, so having every part is reported
+        // there rather than here.
         if (haveTransactions_ && haveState_)
-        {
-            complete_ = true;
             done();
-        }
     }
 }
 

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

@@ -173,8 +173,10 @@ public:
         {
             peer->charge(resource::kFeeInvalidData, "ledger_data invalid");
         }
-        else if (!san.isUseful())
+        else if (!san.isGood())
         {
+            // Good rather than useful: a batch of nodes we already hold is what an honest second
+            // responder to trigger()'s fan-out sends.
             peer->charge(resource::kFeeUselessData, "ledger_data useless");
         }
     }

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

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

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

@@ -52,7 +52,6 @@ void
 TransactionAcquire::done()
 {
     // mtx_ is held, so this may only post real work rather than do it.
-
     if (failed_)
     {
         JLOG(journal_.debug()) << "Failed to acquire TX set " << hash_;
@@ -185,8 +184,25 @@ TransactionAcquire::takeNodes(
     std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
     std::shared_ptr<Peer> const& peer)
 {
-    ScopedLockType const sl(mtx_);
+    ScopedLockType sl(mtx_);
 
+    auto const san = takeNodesLocked(std::move(data), peer, sl);
+
+    // A batch that advanced the map must keep the next timer tick from counting a timeout against
+    // it. A duplicate counts as an answer: an honest second responder to trigger()'s fan-out has
+    // replied, so no timeout is owed.
+    if (san.isUseful() || san.getDuplicate() > 0)
+        progress_ = true;
+
+    return san;
+}
+
+SHAMapAddNode
+TransactionAcquire::takeNodesLocked(
+    std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
+    std::shared_ptr<Peer> const& peer,
+    ScopedLockType&)
+{
     if (complete_)
     {
         JLOG(journal_.trace()) << "TX set complete";
@@ -199,6 +215,10 @@ TransactionAcquire::takeNodes(
         return SHAMapAddNode();
     }
 
+    // Accumulated across the batch, so a packet ending in one bad node still counts the nodes
+    // hooked in ahead of it, as InboundLedger::receiveNode() already does.
+    SHAMapAddNode san;
+
     try
     {
         if (data.empty())
@@ -213,36 +233,45 @@ TransactionAcquire::takeNodes(
                 if (haveRoot_)
                 {
                     JLOG(journal_.debug()) << "Got root TXS node, already have it";
+                    san.incDuplicate();
+                    continue;
                 }
-                else if (!map_->addRootNode(SHAMapHash{hash_}, std::move(d.second), nullptr)
-                              .isGood())
+
+                auto const result =
+                    map_->addRootNode(SHAMapHash{hash_}, std::move(d.second), nullptr);
+                san += result;
+
+                if (!result.isGood())
                 {
                     JLOG(journal_.warn()) << "TX acquire got bad root node for TX set " << hash_
                                           << " from peer " << peer->id();
-                    return SHAMapAddNode::invalid();
-                }
-                else
-                {
-                    haveRoot_ = true;
+                    return san;
                 }
+
+                haveRoot_ = true;
+                continue;
             }
-            else if (!map_->addKnownNode(d.first, std::move(d.second), &sf).isGood())
+
+            auto const result = map_->addKnownNode(d.first, std::move(d.second), &sf);
+            san += result;
+
+            if (!result.isGood())
             {
                 JLOG(journal_.warn()) << "TX acquire got bad non-root node " << d.first
                                       << " for TX set " << hash_ << " from peer " << peer->id();
-                return SHAMapAddNode::invalid();
+                return san;
             }
         }
 
         trigger(peer);
-        progress_ = true;
-        return SHAMapAddNode::useful();
+        return san;
     }
     catch (std::exception const& ex)
     {
         JLOG(journal_.error()) << "Peer " << peer->id()
                                << " sent us junky transaction node data: " << ex.what();
-        return SHAMapAddNode::invalid();
+        san.incInvalid();
+        return san;
     }
 }
 
@@ -268,10 +297,22 @@ TransactionAcquire::init(int numPeers)
 void
 TransactionAcquire::stillNeed()
 {
-    ScopedLockType const sl(mtx_);
+    ScopedLockType sl(mtx_);
 
     timeouts_ = std::min<int>(timeouts_, kNormTimeouts);
+
+    // A running acquisition keeps the wait it has, rather than restarting it for every consensus
+    // round that asks for the set again.
+    if (!failed_)
+        return;
+
     failed_ = false;
+
+    // Restarting the timer is what resumes the acquisition. expires_after() cancels whatever wait
+    // was outstanding, so the timer holds at most one wait at a time. A job queueJob() already
+    // handed to the JobQueue is not canceled by that and still runs one invokeOnTimer(), which
+    // re-arms this same timer and so folds back into the one chain.
+    setTimer(sl);
 }
 
 }  // namespace xrpl

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

@@ -48,6 +48,15 @@ public:
         std::chrono::milliseconds retryInterval = kRetryInterval);
     ~TransactionAcquire() override = default;
 
+    /**
+     * Add nodes a peer sent us to the set we are acquiring.
+     *
+     * @param data The nodes to add, each with its claimed position.
+     * @param peer The peer that sent them.
+     * @return The tally of useful, duplicate, and bad nodes in the batch. Useful and
+     *         bad can both be nonzero, since only the node the batch stops on is
+     *         bad.
+     */
     SHAMapAddNode
     takeNodes(
         std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
@@ -56,6 +65,13 @@ public:
     void
     init(int startPeers);
 
+    /**
+     * Resume a timed-out acquisition, or leave a running one alone.
+     *
+     * Always clamps the timeout count. An acquisition that failed has its timer
+     * chain stopped, so this also clears the failed flag and restarts the timer;
+     * one that is still running already has a timer pending.
+     */
     void
     stillNeed();
 
@@ -68,6 +84,23 @@ private:
     bool haveRoot_{false};
     std::unique_ptr<PeerSet> peerSet_;
 
+    /**
+     * Add nodes a peer sent us, on the lock takeNodes() holds.
+     *
+     * Split out so recording what the batch achieved happens on one exit rather
+     * than on each of the several this has, including the ones that stop the batch
+     * early.
+     *
+     * @param data The nodes to add, each with its claimed position.
+     * @param peer The peer that sent them.
+     * @return The tally of useful, duplicate, and bad nodes in the batch.
+     */
+    SHAMapAddNode
+    takeNodesLocked(
+        std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
+        std::shared_ptr<Peer> const& peer,
+        ScopedLockType&);
+
     void
     onTimer(bool progress, ScopedLockType& sl) override;