fix: Bound a late TX-set reply, and turn it away before parsing it

src/test/app/TransactionAcquire_test.cpp

@@ -261,6 +261,116 @@ struct TransactionAcquire_test : public beast::unit_test::Suite
             BEAST_EXPECT(delivered->getHash() == chain.rootHash);
     }
 
+    /**
+     * The late-reply allowance must survive giveSet(), through the real
+     * dispatch rather than by calling takeNodes() directly.
+     *
+     * giveSet() only resets the acquisition when something else supplied
+     * the set (see the fromAcquire guard), so a reply arriving after
+     * completion still reaches getAcquire() and takeNodesLocked()'s
+     * allowance instead of gotData()'s unconditional ta == nullptr charge.
+     * Driven end to end through InboundTransactions::gotData(), including
+     * the async job done() hands off to, since every other late-reply test
+     * in this suite calls takeNodes() directly and so never exercises that
+     * path.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testLateReplyAllowanceSurvivesGiveSet(jtx::Env& env)
+    {
+        testcase("The late-reply allowance survives giveSet()");
+
+        auto const chain = DeepChain::toLeaf(3, nextSeed());
+        auto& inbound = env.app().getInboundTransactions();
+
+        uint256 const setHash = chain.rootHash.asUInt256();
+        BEAST_EXPECT(inbound.getSet(setHash, true) == nullptr);
+
+        // One peer supplies the whole chain, so it alone earns the one targeted follow-up
+        // request the root reply buys, and so the allowance's one slot.
+        auto const rootPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, rootPeer, packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}}));
+        BEAST_EXPECT(rootPeer->charges().empty());
+
+        inbound.gotData(setHash, rootPeer, packetFor(chain, chain.nodesBelowRoot()));
+        BEAST_EXPECT(rootPeer->charges().empty());
+
+        // Proves the acquisition completed and handed off through the real done()/giveSet()
+        // path, rather than this case racing takeNodes() directly the way the rest of the
+        // suite does.
+        auto const delivered = waitForDeliveredSet(env, setHash);
+        BEAST_EXPECT(delivered != nullptr);
+
+        // giveSet() keeps the acquisition registered, so gotData() finds it here and the
+        // allowance in takeNodesLocked() runs. This assertion is what pins that registration.
+        // rootPeer's own late reply is the one whose slot this is, and is free.
+        inbound.gotData(setHash, rootPeer, packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}}));
+        BEAST_EXPECT(rootPeer->charges().empty());
+
+        // A second reply from rootPeer has already spent that slot: this one is a replay.
+        inbound.gotData(setHash, rootPeer, packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}}));
+        BEAST_EXPECT(rootPeer->charges() == std::vector{resource::kFeeUselessData});
+    }
+
+    /**
+     * A late reply is turned away before its nodes are deserialized.
+     *
+     * Observed through the fee tier, which is what the order is visible
+     * in: unparseable node data charges kFeeInvalidData when parsed, and
+     * nothing or kFeeUselessData when turned away first. See
+     * testUndeserializableNodeIsCharged, which feeds the same packet to a
+     * running acquisition and does get kFeeInvalidData.
+     *
+     * @param env The environment to run in.
+     */
+    void
+    testLateReplyIsTurnedAwayBeforeParsing(jtx::Env& env)
+    {
+        testcase("A late reply is turned away before its nodes are parsed");
+
+        auto const chain = DeepChain::toLeaf(3, nextSeed());
+        auto& inbound = env.app().getInboundTransactions();
+
+        uint256 const setHash = chain.rootHash.asUInt256();
+        BEAST_EXPECT(inbound.getSet(setHash, true) == nullptr);
+
+        // One peer supplies the whole chain, so it alone holds the allowance's one slot. Delivered
+        // in two replies rather than one, since only a peer we asked holds a slot at all, and it is
+        // the follow-up request the root-only reply provokes that puts this one in requestedPeers_.
+        // A single reply completing the set instead reaches trigger()'s no-nodes-missing exit,
+        // which settles the acquisition without ever asking its supplier for anything.
+        auto const rootPeer = std::make_shared<ChargeRecordingPeer>();
+        inbound.gotData(setHash, rootPeer, packetFor(chain, {{SHAMapNodeID{}, chain.nodeAt(0)}}));
+        inbound.gotData(setHash, rootPeer, packetFor(chain, chain.nodesBelowRoot()));
+
+        BEAST_EXPECT(waitForDeliveredSet(env, setHash) != nullptr);
+        BEAST_EXPECT(rootPeer->charges().empty());
+
+        // A single byte naming a wire type that does not exist, as in
+        // testUndeserializableNodeIsCharged.
+        auto const garbage = [&] {
+            auto packet = std::make_shared<protocol::TMLedgerData>();
+            packet->set_ledgerhash(setHash.data(), uint256::size());
+            packet->set_ledgerseq(0);
+            packet->set_type(protocol::liTS_CANDIDATE);
+
+            auto* const node = packet->add_nodes();
+            node->set_nodedata("\xff", 1);
+            node->set_id(SHAMapNodeID{}.getRawString());
+            return packet;
+        };
+
+        // Free, and turned away before the parse, which is what the kFeeUselessData tier shows:
+        // a parsed byte of this shape charges kFeeInvalidData.
+        inbound.gotData(setHash, rootPeer, garbage());
+        BEAST_EXPECT(rootPeer->charges().empty());
+
+        // The slot is spent, so this one is charged as a replay rather than as bad data.
+        inbound.gotData(setHash, rootPeer, garbage());
+        BEAST_EXPECT(rootPeer->charges() == std::vector{resource::kFeeUselessData});
+    }
+
     /**
      * A chain reaching kLeafDepth must end the acquisition outright.
      *
@@ -1127,6 +1237,8 @@ struct TransactionAcquire_test : public beast::unit_test::Suite
 
         testHappyPathCompletesAcquisition(env);
         testTwoPeersEachSupplyPartOfTheSet(env);
+        testLateReplyAllowanceSurvivesGiveSet(env);
+        testLateReplyIsTurnedAwayBeforeParsing(env);
         testFabricatedChainFailsAcquire(env);
         testWrongNodeKeepsAcquireAlive(env);
         testBadRootKeepsAcquireAlive(env);

src/xrpld/app/ledger/InboundTransactions.h

@@ -52,7 +52,8 @@ class InboundTransactions
      * Add a transaction set from a LedgerData message.
      *
      * @param setHash The transaction set ID (digest of the SHAMap root node).
-     * @param peer The peer that sent the message.
+     * @param peer The peer that sent the message, charged here for a reply
+     *        outside its allowance.
      * @param message The LedgerData message.
      */
     virtual void
@@ -66,11 +67,13 @@ class InboundTransactions
      *
      * @param setHash The transaction set ID (should match set.getHash()).
      * @param set The transaction set.
-     * @param acquired Whether this transaction set was acquired from a peer,
-     * or constructed by ourself during consensus.
+     * @param fromAcquire Whether the acquisition for this hash supplied the set
+     *        itself. False cancels an acquisition still in flight; true leaves
+     *        it registered until newRound() sweeps it, so a late reply for the
+     *        hash still reaches takeNodesLocked()'s allowance.
      */
     virtual void
-    giveSet(uint256 const& setHash, std::shared_ptr<SHAMap> const& set, bool acquired) = 0;
+    giveSet(uint256 const& setHash, std::shared_ptr<SHAMap> const& set, bool fromAcquire) = 0;
 
     /**
      * Informs the container if a new consensus round

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

@@ -142,6 +142,11 @@ class InboundTransactionsImp : public InboundTransactions
             return;
         }
 
+        // Asked before the loop below, which costs a hash per node, since a settled acquisition
+        // discards the result. Charges the peer itself when the reply is outside its allowance.
+        if (!ta->wantsReplyFrom(peer))
+            return;
+
         std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data;
         data.reserve(packet.nodes().size());
 
@@ -193,7 +198,12 @@ class InboundTransactionsImp : public InboundTransactions
                 inboundSet.set = set;
             }
 
-            inboundSet.acquire.reset();
+            // Reset only when something other than the acquisition itself supplied the set:
+            // dropping the pointer cancels an acquisition still in flight. Keeping it until
+            // newRound() sweeps the entry lets a late reply for this hash still reach
+            // takeNodesLocked()'s allowance.
+            if (!fromAcquire)
+                inboundSet.acquire.reset();
         }
 
         if (isNew)

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

@@ -217,21 +217,29 @@ TransactionAcquire::takeNodes(
     return san;
 }
 
+bool
+TransactionAcquire::wantsReplyFrom(std::shared_ptr<Peer> const& peer)
+{
+    ScopedLockType sl(mtx_);
+
+    if (!isDone())
+        return true;
+
+    JLOG(journal_.trace()) << (complete_ ? "TX set complete" : "TX set failed");
+    chargeLateReply(peer, sl);
+    return false;
+}
+
 SHAMapAddNode
 TransactionAcquire::takeNodesLocked(
     std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
     std::shared_ptr<Peer> const& peer,
     ScopedLockType& sl)
 {
-    // A reply that arrives after the set is settled - by completing it, or by a different packet
-    // failing it. trigger() sends to every peer it was given, so any of their replies, including
-    // another packet from the same peer whose data failed the set, can already be in flight and
-    // could not have known the outcome. Those are solicited, and free: one per peer we asked,
-    // which is what bounds the honest case.
-    //
-    // Past that bound, further data for this hash is a replay - a resend of data already
-    // accepted or now known worthless, not a first-time reply - and serving it is not free
-    // work, so it is charged.
+    // A reply that arrives after the set is settled, either from a caller that reached
+    // takeNodes() directly or from one whose set settled after wantsReplyFrom() answered.
+    // wantsReplyFrom() returns early for the replies it charges, so this one's allowance is
+    // still unspent.
     if (isDone())
     {
         JLOG(journal_.trace()) << (complete_ ? "TX set complete" : "TX set failed");

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

@@ -72,6 +72,23 @@ class TransactionAcquire : public TimeoutCounter,
         std::vector<std::pair<SHAMapNodeID, SHAMapTreeNodePtr>> data,
         std::shared_ptr<Peer> const& peer);
 
+    /**
+     * Whether a reply from this peer is worth deserializing.
+     *
+     * A settled acquisition discards whatever a reply contains, so it spends the
+     * late-reply allowance here, before the caller turns a wire packet into
+     * nodes at a hash apiece.
+     *
+     * The answer can go stale, since mtx_ is released before takeNodes(), which
+     * recognizes a late reply of its own accord.
+     *
+     * @param peer The peer that sent the reply, charged here when the reply is
+     *        outside the allowance.
+     * @return Whether the reply should be parsed and handed to takeNodes().
+     */
+    [[nodiscard]] bool
+    wantsReplyFrom(std::shared_ptr<Peer> const& peer);
+
     void
     init(int startPeers);
 
@@ -146,8 +163,9 @@ class TransactionAcquire : public TimeoutCounter,
     /**
      * Spend this peer's one free late reply, or charge it for replaying.
      *
-     * Called from takeNodesLocked(), which is where a late reply is
-     * recognized.
+     * Shared by wantsReplyFrom() and takeNodesLocked(), the two places a late
+     * reply is recognized. Only one of them sees any given reply, so a reply is
+     * charged once.
      *
      * @param peer The peer that sent the reply.
      * @param sl Proof mtx_ is held, which the allowance sets require.