rippled/src/test/app/InboundLedger_test.cpp

#include <test/app/AcquireTestHelpers.h>
#include <test/jtx/Env.h>

#include <xrpld/app/ledger/InboundLedger.h>
#include <xrpld/app/ledger/InboundLedgers.h>
#include <xrpld/app/ledger/LedgerMaster.h>

#include <xrpl/basics/Blob.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/unit_test/suite.h>
#include <xrpl/ledger/Ledger.h>
#include <xrpl/nodestore/NodeObject.h>
#include <xrpl/protocol/HashPrefix.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/Serializer.h>
#include <xrpl/protocol/jss.h>
#include <xrpl/resource/Fees.h>
#include <xrpl/shamap/SHAMapNodeID.h>
#include <xrpl/shamap/SHAMapTreeNode.h>

#include <xrpl.pb.h>

#include <chrono>
#include <memory>
#include <mutex>
#include <set>
#include <utility>
#include <vector>

namespace xrpl::test {

/**
 * An acquisition that exposes the entry points its bases keep protected, so a
 * case can reach them through this subclass alone.
 */
struct TestableInboundLedger final : InboundLedger
{
    using InboundLedger::InboundLedger;

    /**
     * Look for the ledger locally, ask the peers being tracked for the rest,
     * and arm the timer, as InboundLedgers::acquire() does. That caller holds
     * its collection lock across init(), which releases it, so this stands in
     * with a lock of its own. The mutex is declared first, so it outlives the
     * lock.
     */
    void
    startAcquire()
    {
        std::recursive_mutex collectionMutex;
        ScopedLockType collectionLock(collectionMutex);
        init(collectionLock);
    }

    /**
     * Hold the acquisition's own lock. The timer chain takes this lock in
     * invokeOnTimer(), so holding it keeps the chain from running until the
     * lock is released. The lock is recursive, so init() can take it again.
     *
     * @return The held lock.
     */
    [[nodiscard]] ScopedLockType
    lock() const
    {
        return ScopedLockType(mtx_);
    }

    /**
     * Ask for more nodes, or judge what has been collected, as a fresh
     * acquisition does.
     */
    void
    triggerAdded()
    {
        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;
    }

    /**
     * Whether a packet has advanced the acquisition since the flag was last
     * cleared.
     *
     * @return Whether progress has been recorded.
     */
    [[nodiscard]] bool
    madeProgress() const
    {
        ScopedLockType const sl(mtx_);
        return progress_;
    }

    /**
     * Record that every part has been fetched.
     */
    void
    markComplete()
    {
        ScopedLockType const sl(mtx_);
        complete_ = true;
    }

    /**
     * Settle the acquisition and signal whatever is waiting on it.
     */
    void
    signalDone()
    {
        ScopedLockType const sl(mtx_);
        done();
    }
};

/**
 * The ledger an acquisition is assembling, as a pointer that can modify it.
 *
 * @param acquire The acquisition to read from.
 * @return The ledger, or nullptr if there is none to report.
 */
[[nodiscard]] static std::shared_ptr<Ledger>
mutableLedger(InboundLedger const& acquire)
{
    // Sound because the acquisition holds a non-const ledger and only hands out a const view.
    return std::const_pointer_cast<Ledger>(acquire.getLedger());
}

struct InboundLedger_test : public beast::unit_test::Suite
{
    /**
     * A retry interval short enough that a whole timeout chain costs a fraction
     * of a second. TimeoutCounter refuses anything at or below 10ms.
     */
    static constexpr auto kFastRetry = std::chrono::milliseconds{20};

    /**
     * A seed unique to this chain within the suite.
     *
     * The Env below is shared, and its node store, fetch packs and remembered
     * failures are all keyed by hash. A fresh seed per chain gives every chain
     * distinct hashes, so each case resolves only its own nodes.
     *
     * @return The seed.
     */
    [[nodiscard]] unsigned int
    nextSeed()
    {
        return ++seed_;
    }

    /**
     * A ledger header naming the given map roots, with a hash derived from the
     * fields, so an acquisition accepts the header as its own however the roots
     * are chosen.
     *
     * @param txHash The transaction map root. Zero means no transactions.
     * @param accountHash The state map root. Zero is a ledger no acquisition
     *        can finish.
     * @return The header, with its hash filled in.
     */
    static LedgerHeader
    makeHeader(UInt256 const& txHash, UInt256 const& accountHash)
    {
        LedgerHeader header;
        header.seq = 2;
        header.parentCloseTime = NetClock::time_point{};
        header.closeTime = NetClock::time_point{};
        header.closeTimeResolution = NetClock::duration{10};
        header.closeFlags = 0;
        header.txHash = txHash;
        header.accountHash = accountHash;
        header.hash = calculateLedgerHash(header);
        return header;
    }

    /**
     * The common shape: no transactions, so only the state map is in play.
     *
     * @param chain The chain whose root to name as the state hash.
     * @return The header, with its hash filled in.
     */
    static LedgerHeader
    makeHeader(DeepChain const& chain)
    {
        return makeHeader(UInt256{}, chain.rootHash.asUInt256());
    }

    /**
     * Put the header in the local store, which is the first place tryDB()
     * looks. The store keeps its entries, where a fetch pack hands each out
     * once, so more than one acquisition of the same ledger can find it.
     *
     * @param env The environment whose node store to seed.
     * @param header The header to store, keyed by its own hash.
     */
    static void
    storeHeader(jtx::Env& env, LedgerHeader const& header)
    {
        Serializer s;
        s.add32(HashPrefix::LedgerMaster);
        addRaw(header, s);

        env.app().getNodeFamily().db().store(
            NodeObjectType::Ledger, std::move(s.modData()), header.hash, header.seq);
    }

    /**
     * Put every node of a chain in the local store, so a state-map walk
     * resolves the whole map from the store.
     *
     * @param env The environment whose node store to seed.
     * @param header The header whose sequence the nodes are stored under.
     * @param chain The chain supplying the nodes.
     * @param maxDepth The deepest node to store, so a caller can leave a walk
     *        something to ask for.
     */
    static void
    storeStateNodes(
        jtx::Env& env,
        LedgerHeader const& header,
        DeepChain const& chain,
        unsigned int maxDepth)
    {
        auto& db = env.app().getNodeFamily().db();

        for (auto depth = 0u; depth <= maxDepth; ++depth)
        {
            db.store(
                NodeObjectType::AccountNode,
                chain.prefixedNodeAt(depth),
                chain.nodeAt(depth)->getHash().asUInt256(),
                header.seq);
        }
    }

    /**
     * The header alone as a liBASE reply, so takeHeader() alone judges the
     * packet.
     *
     * @param header The header to carry, serialized without a hash prefix, as
     *        takeHeader() expects.
     * @return The reply packet.
     */
    static std::shared_ptr<protocol::TMLedgerData>
    headerPacket(LedgerHeader const& header)
    {
        Serializer s;
        addRaw(header, s);

        auto packet = std::make_shared<protocol::TMLedgerData>();
        packet->set_ledgerhash(header.hash.data(), UInt256::size());
        packet->set_ledgerseq(header.seq);
        packet->set_type(protocol::liBASE);
        packet->add_nodes()->set_nodedata(s.peekData().data(), s.peekData().size());
        return packet;
    }

    /**
     * An acquisition started with nothing in the store for its hash, so it
     * waits on its peers.
     *
     * @param env The environment to run in.
     * @param header The header of the ledger to acquire.
     * @return The acquisition, started.
     */
    static std::shared_ptr<TestableInboundLedger>
    startPeerAcquire(jtx::Env& env, LedgerHeader const& header)
    {
        auto acquire = std::make_shared<TestableInboundLedger>(
            env.app(),
            header.hash,
            header.seq,
            InboundLedger::Reason::GENERIC,
            stopwatch(),
            std::make_unique<RequestCountingPeerSet>());
        acquire->startAcquire();
        return acquire;
    }

    /**
     * The chain's state-map nodes as a liAS_NODE reply for the given header.
     *
     * @param header The header whose hash and sequence the reply names.
     * @param chain The chain supplying the nodes.
     * @param data The nodes to include, each with its claimed position.
     * @return The reply packet.
     */
    static std::shared_ptr<protocol::TMLedgerData>
    stateNodePacket(
        LedgerHeader const& header,
        DeepChain const& chain,
        std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> const& data)
    {
        return packetFor(chain, data, protocol::liAS_NODE, header.hash, header.seq);
    }

    /**
     * A ledger whose maps all resolve locally finishes on the spot, and the
     * finished ledger is immutable and handed on.
     *
     * Both entry points into tryDB() are driven: InboundLedgers::acquire(), the
     * only caller of init(), and checkLocal(), which reaches done().
     *
     * @param env The environment to run in.
     */
    void
    testLocalLedgerCompletesAcquire(jtx::Env& env)
    {
        testcase("A ledger found locally completes the acquire");

        // A chain ending in a real leaf, so the state map is complete rather than merely rooted.
        // No transactions, so only the state map is in play.
        auto const chain = DeepChain::toLeaf(2, nextSeed());
        auto const header = makeHeader(chain);

        storeHeader(env, header);
        storeStateNodes(env, header, chain, chain.deepestDepth);

        // acquire() returns the ledger only once the acquisition is complete and unfailed, so a
        // non-null result shows tryDB() found it in the store.
        auto const acquired = env.app().getInboundLedgers().acquire(
            header.hash, header.seq, InboundLedger::Reason::GENERIC);

        BEAST_EXPECT(acquired != nullptr);
        if (acquired)
        {
            BEAST_EXPECT(acquired->isImmutable());
            BEAST_EXPECT(acquired->header().hash == header.hash);
        }

        // init() hands a ledger it completed to LedgerMaster itself.
        BEAST_EXPECT(env.app().getLedgerMaster().getLedgerByHash(header.hash) != nullptr);

        // Nothing on init()'s path records a failure, since init() never reaches done(). This is
        // the baseline for the check after checkLocal() below: an entry found there came from
        // done().
        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));

        // The reason decides where a settled ledger goes: HISTORY counts it toward the fetch rate
        // instead of handing it to LedgerMaster.
        {
            auto const historyChain = DeepChain::toLeaf(2, nextSeed());
            auto const historyHeader = makeHeader(historyChain);

            storeHeader(env, historyHeader);
            storeStateNodes(env, historyHeader, historyChain, historyChain.deepestDepth);

            // onLedgerFetched() is the only writer of this rate, and done() calls it before it
            // posts any work, so the read below is not racing the job queue.
            auto const rateBefore = env.app().getInboundLedgers().fetchRate();

            auto const history = std::make_shared<InboundLedger>(
                env.app(),
                historyHeader.hash,
                historyHeader.seq,
                InboundLedger::Reason::HISTORY,
                stopwatch(),
                std::make_unique<RequestCountingPeerSet>());

            BEAST_EXPECT(history->checkLocal());
            BEAST_EXPECT(history->isComplete());
            BEAST_EXPECT(!history->isFailed());

            // The switch the rate check reads is reached only once the ledger has been settled.
            auto const historyLedger = history->getLedger();
            BEAST_EXPECT(historyLedger != nullptr);
            if (historyLedger)
                BEAST_EXPECT(historyLedger->isImmutable());

            BEAST_EXPECT(env.app().getInboundLedgers().fetchRate() > rateBefore);
        }

        // The same ledger through checkLocal(), which unlike init() reaches done().
        auto const again = std::make_shared<InboundLedger>(
            env.app(),
            header.hash,
            header.seq,
            InboundLedger::Reason::GENERIC,
            stopwatch(),
            std::make_unique<RequestCountingPeerSet>());

        // True because the acquisition ended, which it reports only after done() has run.
        BEAST_EXPECT(again->checkLocal());
        BEAST_EXPECT(again->isComplete());
        BEAST_EXPECT(!again->isFailed());

        auto const settled = again->getLedger();
        BEAST_EXPECT(settled != nullptr);
        if (settled)
            BEAST_EXPECT(settled->isImmutable());

        // done() records a failure only inside the AcqDone job it queues, and checkLocal() returns
        // before that job runs. Wait for the job, so the list is read after done() has had its say.
        env.app().getJobQueue().rendezvous();
        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.
     *
     * done() settles the ledger before it logs or acts on the outcome, and an
     * abandoned map makes settling refuse, so the acquisition records a
     * failure.
     *
     * @param env The environment to run in.
     */
    void
    testInvalidatedLedgerFailsInDone(jtx::Env& env)
    {
        testcase("A ledger invalidated on its way to being settled fails");

        // The fabricated chain, so feeding it to the state map invalidates the map.
        DeepChain const chain{nextSeed()};

        // Only the header is local, so the acquisition holds a ledger with an empty state map.
        auto const header = makeHeader(chain);
        storeHeader(env, header);

        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->isFailed());
        BEAST_EXPECT(!acquire->isComplete());

        auto const ledger = mutableLedger(*acquire);
        BEAST_EXPECT(ledger != nullptr);
        if (!ledger)
            return;

        // The state of affairs done() is handed: every part fetched, as far as the caller can tell.
        acquire->markComplete();

        // And the walk that has since reached the verdict.
        auto& stateMap = ledger->stateMap();
        BEAST_EXPECT(stateMap.addRootNode(chain.rootHash, chain.nodeAt(0), nullptr).isGood());
        for (auto const& [nodeID, node] : chain.nodesBelowRoot())
            stateMap.addKnownNode(nodeID, node, nullptr);
        BEAST_EXPECT(!stateMap.isValid());

        acquire->signalDone();

        // complete_ is withdrawn alongside the failure, or every guard that checks it before
        // failed_ keeps treating this ledger as delivered.
        BEAST_EXPECT(!acquire->isComplete());
        BEAST_EXPECT(acquire->isFailed());

        // This failure leaves ledger_ set, so only the failed_ guard in getLedger() withholds it.
        BEAST_EXPECT(acquire->getLedger() == nullptr);

        // getLedgerByHash answers null for the hash, and the hash is remembered as a failure,
        // which defers re-acquisition.
        BEAST_EXPECT(env.app().getLedgerMaster().getLedgerByHash(header.hash) == nullptr);
        BEAST_EXPECT(waitFor([&] { return env.app().getInboundLedgers().isFailure(header.hash); }));
    }

    /**
     * A packet for an abandoned map must fail rather than be read as the map
     * finishing.
     *
     * addKnownNode() reports every node a duplicate rather than invalid once
     * the map is no longer Synching (see SHAMap::addKnownNode), so
     * receiveNode() must read isValid() rather than isSynching() alone.
     *
     * The tx hash never resolves, so haveTransactions_ stays false and
     * nothing but the check under test can make this packet report success.
     *
     * @param env The environment to run in.
     */
    void
    testInvalidatedMapFailsReceiveNode(jtx::Env& env)
    {
        testcase("A packet for an abandoned map fails rather than completes");

        // The fabricated chain, so feeding it to the state map invalidates the map.
        DeepChain const chain{nextSeed()};

        auto const header = makeHeader(UInt256{777}, chain.rootHash.asUInt256());
        storeHeader(env, header);

        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->isFailed());
        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());

        // Invalidate the state map directly, standing in for a concurrent trigger() walk that
        // reached the same verdict with mtx_ released.
        auto const ledger = mutableLedger(*acquire);
        BEAST_EXPECT(ledger != nullptr);
        if (!ledger)
            return;

        auto& stateMap = ledger->stateMap();
        BEAST_EXPECT(stateMap.addRootNode(chain.rootHash, chain.nodeAt(0), nullptr).isGood());
        for (auto const& [nodeID, node] : chain.nodesBelowRoot())
            stateMap.addKnownNode(nodeID, node, nullptr);
        BEAST_EXPECT(!stateMap.isValid());

        // An ordinary state-node packet, arriving as if it were already in flight when the
        // invalidation above happened. addKnownNode()'s "while not synching" branch reports it a
        // duplicate, whatever it contains, since the map is no longer Synching.
        auto const latePeer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(
            latePeer, stateNodePacket(header, chain, {{chain.idAt(1), chain.nodeAt(1)}})));
        acquire->runData();

        // The verdict belongs to the concurrent walk, so the sender keeps its fee and the
        // acquisition fails, leaving the duplicate verdict unread as a finished map.
        BEAST_EXPECT(latePeer->charges().empty());
        BEAST_EXPECT(acquire->isFailed());
        BEAST_EXPECT(!acquire->isComplete());

        // This assertion pins the isValid() check, which is what fails the acquisition while the
        // map behind it is broken.
        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());
    }

    /**
     * An acquisition that fails on local data must still signal.
     *
     * Both entry points that reach tryDB() are covered. done() is what
     * signals, runs logFailure(), and lands the hash in recentFailures_,
     * which stops the next round asking for the same ledger.
     * recentFailures_ is what the assertions watch, since it is the
     * caller-visible consequence of having signaled.
     *
     * @param env The environment to run in.
     */
    void
    testLocalFailureSignalsDone(jtx::Env& env)
    {
        testcase("An acquisition that fails locally still signals");

        // A zero account hash is a ledger no acquisition can ever finish, and tryDB() says so as
        // soon as it has the header.
        auto const header = makeHeader(UInt256{}, UInt256{});
        storeHeader(env, header);

        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));

        // acquire() is the only caller of init(), and returns nullptr for a failed acquisition.
        BEAST_EXPECT(
            env.app().getInboundLedgers().acquire(
                header.hash, header.seq, InboundLedger::Reason::GENERIC) == nullptr);

        // The failure reached recentFailures_, which is what stops the next round asking again.
        BEAST_EXPECT(waitFor([&] { return env.app().getInboundLedgers().isFailure(header.hash); }));

        // The other way tryDB() judges a ledger unobtainable: the header it found is not the one
        // asked for. A non-zero account hash, so the route above cannot be what fails this one.
        auto const strayHeader = makeHeader(UInt256{}, UInt256{2});
        storeHeader(env, strayHeader);

        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(strayHeader.hash));

        // acquire() hands the sequence to the acquisition unscreened, so one past the stored
        // header's is enough to make tryDB() reject what it reads.
        auto const wrongSeq = strayHeader.seq + 1;

        BEAST_EXPECT(
            env.app().getInboundLedgers().acquire(
                strayHeader.hash, wrongSeq, InboundLedger::Reason::GENERIC) == nullptr);

        // tryDB() discards the ledger it built before it gives up, so done() runs here with none
        // to report. The failure is recorded all the same, which is what the call is for.
        BEAST_EXPECT(
            waitFor([&] { return env.app().getInboundLedgers().isFailure(strayHeader.hash); }));

        // The other route into tryDB(): a trigger() on an acquisition that has no header yet. A
        // hash of its own, so this drives a fresh entry.
        auto const otherHeader = makeHeader(UInt256{1}, UInt256{});
        storeHeader(env, otherHeader);

        auto const viaTrigger = std::make_shared<TestableInboundLedger>(
            env.app(),
            otherHeader.hash,
            otherHeader.seq,
            InboundLedger::Reason::GENERIC,
            stopwatch(),
            std::make_unique<RequestCountingPeerSet>());

        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(otherHeader.hash));

        viaTrigger->triggerAdded();

        // Failed before any header state was published: no ledger, and have_header false.
        BEAST_EXPECT(viaTrigger->isFailed());
        BEAST_EXPECT(!viaTrigger->isComplete());
        BEAST_EXPECT(viaTrigger->getLedger() == nullptr);
        BEAST_EXPECT(!viaTrigger->getJson(0)[jss::have_header].asBool());
        BEAST_EXPECT(
            waitFor([&] { return env.app().getInboundLedgers().isFailure(otherHeader.hash); }));
    }

    /**
     * An acquisition whose state root names a shape no valid tree can have must
     * fail, and must cost the sender the harsher tier.
     *
     * @param env The environment to run in.
     */
    void
    testFabricatedChainFailsAcquire(jtx::Env& env)
    {
        testcase("A state-map chain reaching kLeafDepth fails the acquire");

        DeepChain const chain{nextSeed()};
        auto const header = makeHeader(chain);

        auto const acquire = std::make_shared<TestableInboundLedger>(
            env.app(),
            header.hash,
            header.seq,
            InboundLedger::Reason::GENERIC,
            stopwatch(),
            std::make_unique<RequestCountingPeerSet>());

        // The header is accepted on its own terms, so the acquisition now chases this hash.
        auto const headerPeer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(headerPeer, headerPacket(header)));
        acquire->runData();
        BEAST_EXPECT(headerPeer->charges().empty());
        BEAST_EXPECT(!acquire->isFailed());

        // The root, then the rest of the chain ending in the inner node at kLeafDepth.
        auto const data = chain.allNodes();

        // The header counted as progress, so clear it to see what the packet below records.
        acquire->clearProgress();

        auto const chainPeer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(chainPeer, stateNodePacket(header, chain, data)));
        acquire->runData();

        // The acquisition is over, and stays over: no peer can satisfy this hash.
        BEAST_EXPECT(acquire->isFailed());
        BEAST_EXPECT(!acquire->isComplete());

        // The nodes ahead of the bad one belong to the same impossible tree, so the packet counts
        // nothing at all.
        BEAST_EXPECT(!acquire->madeProgress());

        // A failed acquisition reports no ledger, though it still holds the partial one it built.
        BEAST_EXPECT(acquire->getLedger() == nullptr);

        BEAST_EXPECT(chainPeer->charges() == std::vector{resource::kFeeMalformedData});

        // getJson() walks the same maps to report what is still needed, and comes back with an
        // empty list, since the walk stops at the abandoned map.
        auto const report = acquire->getJson(0);
        BEAST_EXPECT(report[jss::failed].asBool());
        BEAST_EXPECT(report[jss::have_header].asBool());
        BEAST_EXPECT(!report[jss::have_state].asBool());
        BEAST_EXPECT(report[jss::needed_state_hashes].size() == 0);
    }

    /**
     * A merely-wrong state node must cost the recoverable tier and leave the
     * acquisition alive. The fee split in receiveNode() turns on whether this
     * node proved the map impossible, so a node that cannot be hooked but
     * leaves the map sound is charged the lower tier.
     *
     * @param env The environment to run in.
     */
    void
    testWrongStateNodeKeepsAcquireAlive(jtx::Env& env)
    {
        testcase("A merely-wrong state node leaves the acquire recoverable");

        DeepChain const chain{nextSeed()};
        auto const header = makeHeader(chain);

        auto const acquire = std::make_shared<InboundLedger>(
            env.app(),
            header.hash,
            header.seq,
            InboundLedger::Reason::GENERIC,
            stopwatch(),
            std::make_unique<RequestCountingPeerSet>());

        auto const headerPeer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(headerPeer, headerPacket(header)));
        acquire->runData();
        BEAST_EXPECT(!acquire->isFailed());

        auto const rootPeer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(
            rootPeer, stateNodePacket(header, chain, {{SHAMapNodeID{}, chain.nodeAt(0)}})));
        acquire->runData();
        BEAST_EXPECT(rootPeer->charges().empty());

        // nodeAt(1) is the node the root is missing and its hash matches, but it is labeled as
        // living at depth 2, so it cannot be hooked anywhere.
        auto const wrongPeer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(
            wrongPeer,
            stateNodePacket(header, chain, {{SHAMapNodeID{2, UInt256{}}, chain.nodeAt(1)}})));
        acquire->runData();

        BEAST_EXPECT(wrongPeer->charges() == std::vector{resource::kFeeInvalidData});

        // The map is sound, so the acquisition is still going and still holds its ledger.
        BEAST_EXPECT(!acquire->isFailed());
        BEAST_EXPECT(!acquire->isComplete());
        BEAST_EXPECT(acquire->getLedger() != nullptr);
    }

    /**
     * A ledger assembled from local data must be judged even when only one map
     * is settled.
     *
     * 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(). Here 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() sets neither flag, so it has
     * to reach the verdict itself.
     *
     * @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
     * recentFailures_ is what shows done() ran.
     *
     * @param env The environment to run in.
     */
    void
    testPeerZeroAccountHashFails(jtx::Env& env)
    {
        testcase("A peer header with a zero account hash fails the acquire");

        // A real transaction root and a zero state root. Nothing is stored for this hash, so only a
        // peer can supply the header.
        auto const chain = DeepChain::toLeaf(2, nextSeed());
        auto const header = makeHeader(chain.rootHash.asUInt256(), UInt256{});

        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));

        auto const acquire = startPeerAcquire(env, header);
        BEAST_EXPECT(!acquire->isFailed());
        BEAST_EXPECT(!acquire->getJson(0)[jss::have_header].asBool());

        auto const peer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(peer, headerPacket(header)));
        acquire->runData();

        // Failed, not complete, and the header is dropped, as tryDB() leaves one.
        BEAST_EXPECT(acquire->isFailed());
        BEAST_EXPECT(!acquire->isComplete());
        BEAST_EXPECT(acquire->getLedger() == nullptr);
        BEAST_EXPECT(!acquire->getJson(0)[jss::have_header].asBool());

        // The header is the one asked for, so the sender is not to blame.
        BEAST_EXPECT(peer->charges().empty());

        // done() ran, so the hash reached recentFailures_, which stops the next round asking again.
        BEAST_EXPECT(waitFor([&] { return env.app().getInboundLedgers().isFailure(header.hash); }));
    }

    /**
     * A peer-supplied header with a zero transaction hash is taken, since that
     * is an empty transaction set, and the acquisition finishes once the peer
     * supplies the state map.
     *
     * @param env The environment to run in.
     */
    void
    testPeerHeaderWithoutTransactionsCompletes(jtx::Env& env)
    {
        testcase("A peer header with no transactions completes the acquire");

        // No transactions, and a state map that ends in a real leaf, so the acquisition can finish.
        auto const chain = DeepChain::toLeaf(2, nextSeed());
        auto const header = makeHeader(chain);

        auto const acquire = startPeerAcquire(env, header);

        auto const peer = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(peer, headerPacket(header)));
        acquire->runData();

        // The header is taken, the empty transaction map counts as fetched, and the state map is
        // what remains.
        BEAST_EXPECT(!acquire->isFailed());
        BEAST_EXPECT(!acquire->isComplete());
        auto const json = acquire->getJson(0);
        BEAST_EXPECT(json[jss::have_header].asBool());
        BEAST_EXPECT(json[jss::have_transactions].asBool());
        BEAST_EXPECT(!json[jss::have_state].asBool());

        // The whole state map as state-node replies, root first, which completes the acquisition.
        // The second gotData() answers false, since the first already asked for a dispatch.
        BEAST_EXPECT(acquire->gotData(
            peer,
            packetFor(
                chain,
                {{SHAMapNodeID{}, chain.nodeAt(0)}},
                protocol::liAS_NODE,
                header.hash,
                header.seq)));
        BEAST_EXPECT(!acquire->gotData(
            peer,
            packetFor(
                chain, chain.nodesBelowRoot(), protocol::liAS_NODE, header.hash, header.seq)));
        acquire->runData();

        BEAST_EXPECT(acquire->isComplete());
        BEAST_EXPECT(!acquire->isFailed());
        BEAST_EXPECT(peer->charges().empty());

        auto const settled = acquire->getLedger();
        BEAST_EXPECT(settled != nullptr);
        if (settled)
            BEAST_EXPECT(settled->isImmutable());

        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
    }

    /**
     * The retry timer re-asks, then gives up and signals.
     *
     * 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.
     */
    void
    testTimerRetriesThenGivesUp(jtx::Env& env)
    {
        testcase("The retry timer re-asks, then gives up");

        UInt256 const kUnknownLedger{8};

        // One candidate, which onTimer() re-offers on every tick.
        auto const candidate = std::make_shared<ChargeRecordingPeer>();
        auto peerSet =
            std::make_unique<RequestCountingPeerSet>(std::vector<std::shared_ptr<Peer>>{candidate});
        auto const* const peerSetPtr = peerSet.get();

        auto acquire = std::make_shared<TestableInboundLedger>(
            env.app(),
            kUnknownLedger,
            0,
            InboundLedger::Reason::GENERIC,
            stopwatch(),
            std::move(peerSet),
            kFastRetry);

        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(kUnknownLedger));

        // init() finds the hash absent from the store, so it asks the candidate and queues the
        // first tick as a job at once rather than arming the timer. The case counts from here, past
        // those first requests. The lock keeps the chain from running until the baseline is
        // recorded. Without it the whole chain (kLedgerTimeoutRetriesMax + 1 ticks, the later ones
        // at kFastRetry) can finish first, and the count below would have nothing to climb from.
        int requestsFromInit = 0;
        {
            auto const held = acquire->lock();
            acquire->startAcquire();
            BEAST_EXPECT(!acquire->isFailed());
            requestsFromInit = peerSetPtr->requests();
        }
        BEAST_EXPECT(requestsFromInit > 0);
        BEAST_EXPECT(peerSetPtr->addedPeers() == std::set<Peer::ID>{candidate->id()});

        // Every tick asks again, and past kLedgerTimeoutRetriesMax (6) the chain gives up.
        BEAST_EXPECT(waitFor([&] { return acquire->isFailed(); }));
        BEAST_EXPECT(!acquire->isComplete());
        BEAST_EXPECT(peerSetPtr->requests() > requestsFromInit);

        // No ledger was ever built here, so both arms of the failed_ gate answer null. The
        // local-chain case above covers the gate itself.
        BEAST_EXPECT(acquire->getLedger() == nullptr);
        BEAST_EXPECT(
            waitFor([&] { return env.app().getInboundLedgers().isFailure(kUnknownLedger); }));
    }

    /**
     * A packet that leaves nothing to fetch must settle the ledger itself,
     * rather than count on a later walk to notice.
     *
     * A map holding one item has that item's leaf for its root, so rooting it
     * stops it synching on the spot. The packet here offers that root to a map
     * that already holds it, which counts no good nodes, so runData() takes the
     * sender for one that returned nothing and asks it nothing further. That
     * leaves receiveNode() as the only thing that can settle this ledger.
     *
     * @param env The environment to run in.
     */
    void
    testCompletesWhenNothingIsLeftToFetch(jtx::Env& env)
    {
        testcase("A packet that leaves nothing to fetch settles the ledger");

        // A leaf at the root, so the state map holds exactly one item. No transactions, which
        // leaves the state map the only part the acquisition needs.
        auto const chain = DeepChain::toLeaf(0, nextSeed());
        auto const header = makeHeader(chain);

        // Only the header is local, so the state map starts with no root of its own.
        storeHeader(env, header);

        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->isComplete());
        BEAST_EXPECT(!acquire->isFailed());
        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());

        // Root the map directly, standing in for a trigger() walk that got that far. The flag is
        // the acquisition's own, so it stays clear while the map is already whole.
        auto const ledger = mutableLedger(*acquire);
        BEAST_EXPECT(ledger != nullptr);
        if (!ledger)
            return;

        auto& stateMap = ledger->stateMap();
        BEAST_EXPECT(stateMap.addRootNode(chain.rootHash, chain.nodeAt(0), nullptr).isUseful());
        BEAST_EXPECT(!stateMap.isSynching());
        BEAST_EXPECT(!acquire->getJson(0)[jss::have_state].asBool());

        // The same root again. A leaf carries its own key, so a depth of zero names the root and
        // the node goes through addRootNode(), which reports a node already held.
        auto const sender = std::make_shared<ChargeRecordingPeer>();
        BEAST_EXPECT(acquire->gotData(sender, stateNodePacket(header, chain, chain.allNodes())));
        acquire->runData();

        // Nothing was asked of the sender afterwards, so no walk ran to reach the same verdict.
        BEAST_EXPECT(!acquire->madeProgress());
        BEAST_EXPECT(peerSetPtr->requests() == 0);

        // The sender offered nothing bad, so it keeps its fee, and the map is now reported held.
        BEAST_EXPECT(sender->charges().empty());
        BEAST_EXPECT(acquire->getJson(0)[jss::have_state].asBool());

        // Settled, reported complete and handed on, all from this one packet.
        BEAST_EXPECT(acquire->isComplete());
        BEAST_EXPECT(!acquire->isFailed());
        BEAST_EXPECT(ledger->isImmutable());

        BEAST_EXPECT(env.app().getLedgerMaster().getLedgerByHash(header.hash) != nullptr);
        BEAST_EXPECT(!env.app().getInboundLedgers().isFailure(header.hash));
    }

    void
    run() override
    {
        // One Env for the suite, shared safely because every chain is seeded through nextSeed().
        jtx::Env env{*this};

        testLocalLedgerCompletesAcquire(env);
        testWalkSettlesBeforeReportingComplete(env);
        testInvalidatedLedgerFailsInDone(env);
        testInvalidatedMapFailsReceiveNode(env);
        testLocalFailureSignalsDone(env);
        testPeerZeroAccountHashFails(env);
        testPeerHeaderWithoutTransactionsCompletes(env);
        testFabricatedChainFailsAcquire(env);
        testWrongStateNodeKeepsAcquireAlive(env);
        testLocalChainFailsAcquire(env);
        testAggressiveRetryJudgesLocalMap(env);
        testWalkJudgesMapOnOrdinaryTrigger(env);
        testCompletesWhenNothingIsLeftToFetch(env);

        // Last: the only case that waits out a whole timeout chain.
        testTimerRetriesThenGivesUp(env);
    }

private:
    unsigned int seed_{0};
};

BEAST_DEFINE_TESTSUITE(InboundLedger, app, xrpl);

}  // namespace xrpl::test