fix: Start a ledger built from a header mutable

Ledger(LedgerHeader const&, Rules, Family&) starts at immutable_ false.
Its maps are constructed Synching and filled in afterwards by an
acquisition or a replay, so setImmutable() is what settles the ledger,
and only once both maps are sound. mapHashesFromHeader_ records that
this constructor's transaction and account hashes are input rather than
derived, so setImmutable() leaves them alone: deriving them from the
maps would relabel a map that fell short of its target. The flag is
const, unlike immutable_.

A ledger this constructor built is therefore unsettled until
setImmutable() says otherwise, and LedgerMaster::switchLCL() and
LedgerHistory::insert() each require a settled one.
InboundLedger::getLedger() already reports nothing once the acquisition
has failed, so done()'s posted job reads the accessor: failure and a
null ledger_ are refused by one check, and checkAccept() requires a
non-null one. isFailed()'s comment said the opposite of what it returns.

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/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/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};
     /**