fix: Settle an acquired ledger before reporting it complete

done() owns the publication of complete_ and sets it only once the
ledger is settled, so a caller reading isComplete() without mtx_ may use
the ledger directly. LedgerHistory::insert(), LedgerHolder::set() and
LedgerMaster::switchLCL() each require an immutable ledger, and that
ordering is what their requirement rests on. trigger() and receiveNode()
set the have-flags and leave the verdict to done(), while tryDB()
settles its own result and sets complete_ itself, so that path arrives
with only the reporting left.

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,246 @@ 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() without any peer data having arrived. It reports true
+        // only because the acquisition ended, which is what this case is about.
+        BEAST_EXPECT(acquire->checkLocal());
+
+        BEAST_EXPECT(acquire->isFailed());
+        BEAST_EXPECT(!acquire->isComplete());
+    }
+
+    /**
+     * 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 +930,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.
      */
@@ -676,10 +995,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/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
@@ -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:
@@ -553,7 +575,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 +625,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 +692,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 +778,6 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
                 else
                 {
                     haveTransactions_ = true;
-
-                    if (haveState_)
-                        complete_ = true;
                 }
             }
             else
@@ -766,12 +802,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 +1026,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/TimeoutCounter.h

@@ -8,6 +8,7 @@
 
 #include <boost/asio/basic_waitable_timer.hpp>
 
+#include <atomic>
 #include <chrono>
 #include <cstdint>
 #include <memory>
@@ -131,10 +132,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};
     /**