fixup: fix tx submission order

When picking TXs to submit to the mempool stop at gaps.
When picking TXs to download use the peer's advertisement order as a
guide, not the TxKey order.

Author: karknu

ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs

@@ -277,18 +277,33 @@ pickSubmitAction PeerActionContext { pacPeerAddr, pacPeerState, pacSharedState }
      else Just txsToSubmit
   where
 
-    -- Filters the unacknowledged txid queue for bodies buffered by this peer
-    -- that are not currently being submitted by another advertiser.
-    -- Returns the list of tx keys ready for immediate submission in the order they
-    -- were originally advertised by the peer.
-    pickBufferedTxsToSubmit =
-      [ txKey
-      | txKey@(TxKey k) <- toList (peerUnacknowledgedTxIds pacPeerState)
-      , IntMap.member k (peerDownloadedTxs pacPeerState)
-      , Just txEntry <- [IntMap.lookup k (sharedTxTable pacSharedState)]
-      , txBufferedByPeer pacPeerAddr txEntry
-      , not (txSubmittingByOther pacPeerAddr txEntry)
-      ]
+    -- Walk the unacknowledged txid queue in peer advertisement order, picking
+    -- bodies buffered by this peer for immediate submission. Stop at the
+    -- first tx that is unresolved and not available from this peer: later
+    -- txs in the peer's stream must not run ahead of earlier ones, otherwise
+    -- a tx may be offered to the mempool before a transaction it depends on
+    -- is confirmed. Txs already resolved elsewhere (present in
+    -- 'sharedRetainedTxs') are skipped over since no further action is
+    -- needed for them.
+    pickBufferedTxsToSubmit = go [] (toList (peerUnacknowledgedTxIds pacPeerState))
+      where
+        go acc [] = reverse acc
+        go acc (txKey@(TxKey k) : rest) =
+          case IntMap.lookup k (sharedTxTable pacSharedState) of
+            Just txEntry
+              | IntMap.member k (peerDownloadedTxs pacPeerState)
+              , txBufferedByPeer pacPeerAddr txEntry
+              , not (txSubmittingByOther pacPeerAddr txEntry) ->
+                  go (txKey : acc) rest
+            _ | retainedMember k (sharedRetainedTxs pacSharedState) ->
+                  -- already resolved via another peer
+                  go acc rest
+            _ | not (IntMap.member k (peerAvailableTxIds pacPeerState))
+              , not (IntMap.member k (peerDownloadedTxs pacPeerState)) ->
+                  -- we have already finished with this tx (previously
+                  -- submitted, or never had a body to submit)
+                  go acc rest
+            _ -> reverse acc
 
 -- | Select transactions to request from the peer, if within policy limits.
 --
@@ -325,18 +340,22 @@ pickRequestTxsAction ctx@PeerActionContext { pacNow, pacPolicy, pacPeerState, pa
 
         leaseUntil = addTime (interTxSpace pacPolicy) pacNow
 
-        -- We pick which TXs to download based on TxKey in ascending order.
-        -- This makes it likely (but not guaranteed) that we end up downloading
-        -- TXs in the order the peer presented them to us.
+        -- Iterate the peer's unacknowledged queue, which preserves the peer's
+        -- advertisement order. Peers are expected to advertise in chain-
+        -- topological order (parents before children), so walking in that
+        -- order aligns fetch order with submission-validity order and a
+        -- child is never requested ahead of its parent when the same peer
+        -- carries both.
         candidates =
           [ (k, txSize)
-          | (k, txSize) <- IntMap.toAscList (peerAvailableTxIds pacPeerState)
+          | TxKey k <- toList (peerUnacknowledgedTxIds pacPeerState)
           , IntSet.notMember k (peerRequestedTxs pacPeerState)
           , IntMap.notMember k (peerDownloadedTxs pacPeerState)
+          , Just txSize <- [IntMap.lookup k (peerAvailableTxIds pacPeerState)]
           ]
 
-        -- Select transactions to request by iterating through candidates in ascending
-        -- key order until the size budget is consumed.
+        -- Select transactions to request by iterating through candidates in
+        -- peer advertisement order until the size budget is consumed.
         go selectedRev selectedSize txTable [] = (reverse selectedRev, selectedSize, txTable)
         go selectedRev selectedSize txTable ((k, txSize) : rest) =
           if exceedsBudget selectedSize txSize

ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs

@@ -1438,6 +1438,7 @@ prop_nextPeerAction_claimsClaimableTx
   -> Property
 prop_nextPeerAction_claimsClaimableTx (Positive peerA0) (Positive peerB0) (Positive peerC0) txid0 txSize0 =
   distinctPeers ==>
+    peerTxLocalStateInvariant peerState0 .&&.
     case peerAction of
          PeerRequestTxs txKeys ->
            conjoin
@@ -1474,8 +1475,11 @@ prop_nextPeerAction_claimsClaimableTx (Positive peerA0) (Positive peerB0) (Posit
           , txAttempts = Map.empty
           }
       }
-    peerState0 = emptyPeerTxLocalState { peerAvailableTxIds = IntMap.singleton k txSize
-                                       , peerScore = peerAScore }
+    peerState0 = emptyPeerTxLocalState
+      { peerUnacknowledgedTxIds = StrictSeq.singleton key
+      , peerAvailableTxIds = IntMap.singleton k txSize
+      , peerScore = peerAScore
+      }
     (peerAction, peerState', sharedState') = nextPeerAction now defaultTxDecisionPolicy peerA peerState0 sharedState0
 
 unit_nextPeerAction_claimsAtScoreDelayThreshold :: (String -> IO ()) -> Assertion
@@ -1510,8 +1514,11 @@ unit_nextPeerAction_claimsAtScoreDelayThreshold step = do
           , txAttempts = Map.empty
           }
       }
-    peerState0 = emptyPeerTxLocalState { peerAvailableTxIds = IntMap.singleton k txSize
-                                        , peerScore = PeerScore 20 now }
+    peerState0 = emptyPeerTxLocalState
+      { peerUnacknowledgedTxIds = StrictSeq.singleton key
+      , peerAvailableTxIds = IntMap.singleton k txSize
+      , peerScore = PeerScore 20 now
+      }
     (peerAction, peerState', sharedState') =
       nextPeerAction now defaultTxDecisionPolicy peeraddr peerState0 sharedState0
 
@@ -1600,6 +1607,7 @@ prop_nextPeerAction_claimsExpiredLease
   -> Property
 prop_nextPeerAction_claimsExpiredLease (Positive oldOwner0) (Positive peerA0) (Positive peerB0) txid0 txSize0 =
   distinctPeers ==>
+    peerTxLocalStateInvariant peerState0 .&&.
     case peerAction of
          PeerRequestTxs txKeys ->
            conjoin
@@ -1636,8 +1644,11 @@ prop_nextPeerAction_claimsExpiredLease (Positive oldOwner0) (Positive peerA0) (P
           , txAttempts = Map.empty
           }
       }
-    peerState0 = emptyPeerTxLocalState { peerAvailableTxIds = IntMap.singleton k txSize
-                                       , peerScore = peerAScore }
+    peerState0 = emptyPeerTxLocalState
+      { peerUnacknowledgedTxIds = StrictSeq.singleton key
+      , peerAvailableTxIds = IntMap.singleton k txSize
+      , peerScore = peerAScore
+      }
     (peerAction, peerState', sharedState') = nextPeerAction now defaultTxDecisionPolicy peerA peerState0 sharedState0
 
 -- Verifies that nextPeerAction still requests an oversized first tx when it
@@ -1648,6 +1659,7 @@ prop_nextPeerAction_requestsOversizedFirstTx
   -> Positive Int
   -> Property
 prop_nextPeerAction_requestsOversizedFirstTx (Positive peeraddr) txid0 (Positive txSize0) =
+  peerTxLocalStateInvariant peerState0 .&&.
   case peerAction of
        PeerRequestTxs [txKey] ->
          conjoin
@@ -1678,7 +1690,8 @@ prop_nextPeerAction_requestsOversizedFirstTx (Positive peeraddr) txid0 (Positive
       , sharedTxTable = IntMap.singleton k (mkTxEntry peeraddr txSize Nothing)
       }
     peerState0 = emptyPeerTxLocalState
-      { peerAvailableTxIds = IntMap.singleton k txSize
+      { peerUnacknowledgedTxIds = StrictSeq.singleton key
+      , peerAvailableTxIds = IntMap.singleton k txSize
       , peerRequestedTxIds = maxNumTxIdsToRequest policy
       }
     (peerAction, peerState', sharedState') = nextPeerAction now policy peeraddr peerState0 sharedState0
@@ -1707,7 +1720,8 @@ unit_nextPeerAction_skipsBlockedAvailableTxs step = do
     kBlocked = 1
     kClaimable = 2
     peerState = emptyPeerTxLocalState
-      { peerAvailableTxIds = IntMap.fromList [(kBlocked, 10), (kClaimable, 11)]
+      { peerUnacknowledgedTxIds = StrictSeq.fromList [blockedKey, claimableKey]
+      , peerAvailableTxIds = IntMap.fromList [(kBlocked, 10), (kClaimable, 11)]
       }
     sharedState :: SharedTxState PeerAddr TxId
     sharedState = emptySharedTxState
@@ -2117,6 +2131,7 @@ prop_nextPeerActionPipelined_secondBodyBatch
   -> Property
 prop_nextPeerActionPipelined_secondBodyBatch (Positive peeraddr) txidA0 txidB0 txSizeA0 txSizeB0 =
   txidA /= txidB ==>
+    peerTxLocalStateInvariant peerState0 .&&.
     case peerAction of
          PeerRequestTxs [txKey] ->
            conjoin
@@ -2141,7 +2156,8 @@ prop_nextPeerActionPipelined_secondBodyBatch (Positive peeraddr) txidA0 txidB0 t
     kA = unTxKey keyA
     kB = unTxKey keyB
     peerState0 = emptyPeerTxLocalState
-      { peerAvailableTxIds = IntMap.singleton kB txSizeB
+      { peerUnacknowledgedTxIds = StrictSeq.fromList [keyA, keyB]
+      , peerAvailableTxIds = IntMap.fromList [(kA, txSizeA), (kB, txSizeB)]
       , peerRequestedTxs = IntSet.singleton kA
       , peerRequestedTxBatches = StrictSeq.singleton (mkRequestedTxBatch [keyA] txSizeA)
       , peerRequestedTxsSize = txSizeA
@@ -2177,6 +2193,7 @@ prop_nextPeerActionPipelined_noThirdBodyBatch
   -> Property
 prop_nextPeerActionPipelined_noThirdBodyBatch (Positive peeraddr) txidA0 txidB0 txidC0 txSizeA0 txSizeB0 txSizeC0 =
   distinctTxIds ==>
+    peerTxLocalStateInvariant peerState0 .&&.
     case peerAction of
          PeerDoNothing _ _ ->
            conjoin
@@ -2201,7 +2218,8 @@ prop_nextPeerActionPipelined_noThirdBodyBatch (Positive peeraddr) txidA0 txidB0
     kB = unTxKey keyB
     kC = unTxKey keyC
     peerState0 = emptyPeerTxLocalState
-      { peerAvailableTxIds = IntMap.singleton kC txSizeC
+      { peerUnacknowledgedTxIds = StrictSeq.fromList [keyA, keyB, keyC]
+      , peerAvailableTxIds = IntMap.fromList [(kA, txSizeA), (kB, txSizeB), (kC, txSizeC)]
       , peerRequestedTxs = IntSet.fromList [kA, kB]
       , peerRequestedTxBatches = StrictSeq.fromList
           [ mkRequestedTxBatch [keyA] txSizeA