fixup: all advertise should be notified

karknu · karknu · commit cddef432bddc · 2026-04-27T10:48:55.000+02:00
Not only idle peers but all peers should be notified when something
changes for one of their TX.s
diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Registry.hs
@@ -203,7 +203,7 @@ withPeer policy TxSubmissionMempoolReader { mempoolGetSnapshot } sharedStateVar
 
     unregisterPeer :: Time -> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
     unregisterPeer now st@SharedTxState { sharedPeers, sharedTxTable, sharedRetainedTxs, sharedTxIdToKey, sharedKeyToTxId, sharedGeneration } =
-      bumpIdlePeerGenerations peersToWake $ st {
+      bumpPeerGenerations peersToWake $ st {
         sharedPeers = sharedPeers',
         sharedTxTable = sharedTxTable',
         sharedRetainedTxs = sharedRetainedTxs,
@@ -544,7 +544,7 @@ updatePeerPhase peeraddr peerPhaseNew
               let sharedPeerState' = sharedPeerState { sharedPeerPhase = peerPhaseNew }
                   st' = st { sharedPeers = Map.insert peeraddr sharedPeerState' sharedPeers
                            , sharedGeneration = sharedGeneration + 1 }
-              in bumpIdlePeerGenerations (phaseWakePeers peerPhaseOld) st'
+              in bumpPeerGenerations (phaseWakePeers peerPhaseOld) st'
             else st
        _ -> st -- TODO error?
   where
diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/State.hs
@@ -220,7 +220,7 @@ applyRequestTxsChoice ctx txsToRequest txsToRequestSize txTable =
         peerRequestedTxsSize = peerRequestedTxsSize (pacPeerState ctx) + txsToRequestSize
       }
     sharedState'' =
-      bumpIdlePeerGenerations
+      bumpPeerGenerations
         (advertisingPeersForTxKeysExcept (pacPeerAddr ctx) requestedKeys (pacSharedState ctx))
         ((pacSharedState ctx) {
           sharedTxTable = txTable,
@@ -952,7 +952,7 @@ handleReceivedTxs mempoolHasTx now policy peeraddr txs peerState sharedState =
 
     -- Flag peers that may now have work available after processing txs.
     sharedState' =
-      bumpIdlePeerGenerations
+      bumpPeerGenerations
         (Set.union receivedWakePeers omittedWakePeers)
         sharedState''
 
@@ -1120,7 +1120,7 @@ handleSubmittedTxs now policy peeraddr acceptedTxs rejectedTxs peerState sharedS
       IntSet.foldl' updateRejected (sharedStateAfterRejectedPeer, Set.empty) rejectedKeys
 
     sharedState' =
-      bumpIdlePeerGenerations
+      bumpPeerGenerations
         (Set.union acceptedAdvertisers rejectedWakePeers)
         sharedState''
 
@@ -1274,7 +1274,7 @@ handleReceivedTxIds mempoolHasTx now policy peeraddr requestedTxIds txidsAndSize
 
     sharedState''
       | sharedChanged || peerAdvertisedKeys' /= peerAdvertisedKeys0 =
-          bumpIdlePeerGenerations peersToWake $
+          bumpPeerGenerations peersToWake $
             sharedStateHandled {
               sharedPeers =
                 if peerAdvertisedKeys' == peerAdvertisedKeys0
diff --git a/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Types.hs b/ouroboros-network/lib/Ouroboros/Network/TxSubmission/Inbound/V2/Types.hs
@@ -59,7 +59,7 @@ module Ouroboros.Network.TxSubmission.Inbound.V2.Types
   , PeerTxLocalState (..)
   , SharedPeerState (..)
   , peerGenerationOf
-  , bumpIdlePeerGenerations
+  , bumpPeerGenerations
     -- TxKey with helper functions
   , TxKey (..)
   , lookupTxKey
@@ -565,22 +565,22 @@ peerGenerationOf peeraddr SharedTxState { sharedPeers } =
          Just SharedPeerState { sharedPeerGeneration } -> sharedPeerGeneration
          Nothing -> error "TxSubmission.V2.peerGenerationOf: missing peer"
 
-bumpIdlePeerGenerations :: Ord peeraddr
-                        => Set.Set peeraddr
-                        -> SharedTxState peeraddr txid
-                        -> SharedTxState peeraddr txid
-bumpIdlePeerGenerations peersToWake st@SharedTxState { sharedPeers } =
+bumpPeerGenerations :: Ord peeraddr
+                    => Set.Set peeraddr
+                    -> SharedTxState peeraddr txid
+                    -> SharedTxState peeraddr txid
+bumpPeerGenerations peersToWake st@SharedTxState { sharedPeers } =
     st {
       sharedPeers = foldl' bumpOne sharedPeers (Set.toList peersToWake)
     }
   where
     bumpOne peersMap peeraddr =
-        Map.adjust bumpIdlePeer peeraddr peersMap
-      where
-        bumpIdlePeer sharedPeerState@SharedPeerState { sharedPeerPhase, sharedPeerGeneration }
-          | sharedPeerPhase == PeerIdle =
-              sharedPeerState { sharedPeerGeneration = sharedPeerGeneration + 1 }
-          | otherwise = sharedPeerState
+      Map.adjust
+        (\sharedPeerState ->
+           sharedPeerState
+             { sharedPeerGeneration = sharedPeerGeneration sharedPeerState + 1 })
+        peeraddr
+        peersMap
 
 lookupTxKey :: HasRawTxId txid
             => txid
diff --git a/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs b/ouroboros-network/tests/lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs
@@ -68,6 +68,7 @@ tests =
     , testProperty "handleSubmittedTxs retains accepted and drops rejected" prop_handleSubmittedTxs_retainsAcceptedAndDropsRejected
     , testProperty "nextPeerAction prioritises submitting buffered owned txs" prop_nextPeerAction_prioritisesSubmit
     , testProperty "nextPeerAction claims claimable tx for best idle advertiser" prop_nextPeerAction_claimsClaimableTx
+    , testProperty "nextPeerAction claims a released tx from another advertiser" prop_nextPeerAction_claimsRejectedTxFromOtherAdvertiser
     , testCaseSteps "nextPeerAction claims a tx once the score delay threshold has elapsed" unit_nextPeerAction_claimsAtScoreDelayThreshold
     , testProperty "nextPeerAction steals expired lease for best idle advertiser" prop_nextPeerAction_claimsExpiredLease
     , testProperty "nextPeerAction requests an oversized first tx within the soft budget" prop_nextPeerAction_requestsOversizedFirstTx
@@ -85,7 +86,7 @@ tests =
     , testProperty "nextPeerAction keeps retained txs before expiry" prop_nextPeerAction_keepsRetained
     , testProperty "PeerDoNothing waits for the earliest shared expiry" prop_nextPeerAction_earliestWakeDelay
     , testProperty "PeerDoNothing uses the current peer generation" prop_nextPeerAction_returnsPeerGeneration
-    , testProperty "handleSubmittedTxs bumps idle advertiser generations" prop_handleSubmittedTxs_bumpsIdleAdvertisers
+    , testProperty "handleSubmittedTxs bumps advertiser generations" prop_handleSubmittedTxs_bumpsAdvertisers
     , testCaseSteps "advertisingPeersForTxExcept scans the authoritative peer key sets" unit_advertisingPeersForTxExcept_scansAuthoritativePeerSets
     , testCaseSteps "removeAdvertisingPeersForResolvedTx clears all advertising peers for a resolved key" unit_removeAdvertisingPeersForResolvedTx_clearsAllAdvertisingPeers
     , testCaseSteps "updatePeerPhase only wakes the peer becoming idle" unit_updatePeerPhase_wakesOnlyBecomingIdlePeer
@@ -746,19 +747,16 @@ prop_handleReceivedTxIds
         case otherPeerOpt of
           Just op | not (null mempoolResolveActiveGroup) ->
             let original = lookupPeerOrFail op sharedStateWithRetained
-                post     = lookupPeerOrFail op sharedState'
-                bumpIfIdle g
-                  | sharedPeerPhase original == PeerIdle = g + 1
-                  | otherwise                            = g
-            in conjoin
+                post     = lookupPeerOrFail op sharedState' in
+            conjoin
                  [ counterexample "other peer's advertised keys not restored"
                      (sharedPeerAdvertisedTxKeys post
                         === sharedPeerAdvertisedTxKeys original)
                  , counterexample "other peer's phase changed unexpectedly"
                      (sharedPeerPhase post === sharedPeerPhase original)
                  , counterexample "other peer's generation bump mismatch"
                      (sharedPeerGeneration post
-                        === bumpIfIdle (sharedPeerGeneration original))
+                        === sharedPeerGeneration original + 1)
                  ]
           _ -> property True
 
@@ -1214,25 +1212,22 @@ prop_handleReceivedTxs
           ]
 
       -- otherPeer is added to wakePeers when any surviving omitted entry
-      -- exists; its generation is bumped iff it was PeerIdle. LateMempool
-      -- is set up single-advertiser so it does not wake otherPeer here.
+      -- exists; its generation is bumped unconditionally. LateMempool is
+      -- set up single-advertiser so it does not wake otherPeer here.
       checkOtherPeerState =
         case otherPeerOpt of
           Just op | not (IntSet.null omittedSurvivingKeys) ->
             let original = lookupPeerOrFail op sharedStateBase
-                post     = lookupPeerOrFail op sharedState'
-                bumpIfIdle g
-                  | sharedPeerPhase original == PeerIdle = g + 1
-                  | otherwise                            = g
-            in conjoin
+                post     = lookupPeerOrFail op sharedState' in
+            conjoin
                  [ counterexample "other peer's advertised keys changed"
                      (sharedPeerAdvertisedTxKeys post
                         === sharedPeerAdvertisedTxKeys original)
                  , counterexample "other peer's phase changed"
                      (sharedPeerPhase post === sharedPeerPhase original)
                  , counterexample "other peer's generation bump mismatch"
                      (sharedPeerGeneration post
-                        === bumpIfIdle (sharedPeerGeneration original))
+                        === sharedPeerGeneration original + 1)
                  ]
           _ -> property True
 
@@ -1520,6 +1515,80 @@ unit_nextPeerAction_claimsAtScoreDelayThreshold step = do
     (peerAction, peerState', sharedState') =
       nextPeerAction now defaultTxDecisionPolicy peeraddr peerState0 sharedState0
 
+-- | A peer whose submission attempt has been cleared (e.g. after a
+-- mempool rejection) must not prevent another advertiser from claiming
+-- the same tx. After one peer's attempt is cleared and its lease
+-- released back to 'TxClaimable', any other peer that still advertises
+-- the tx should be able to claim it on its next 'nextPeerAction' pass.
+--
+-- Exercises the cross-peer retry invariant in the 'txSelectable' /
+-- 'nextPeerAction' path: once no peer has an outstanding attempt on a
+-- tx and its lease is claimable, a still-advertising peer is eligible
+-- to re-claim.
+prop_nextPeerAction_claimsRejectedTxFromOtherAdvertiser :: Property
+prop_nextPeerAction_claimsRejectedTxFromOtherAdvertiser =
+    counterexample "peerB should be able to claim the released tx"
+  $ case action of
+      PeerRequestTxs keys ->
+        counterexample ("keys requested: " ++ show keys)
+          $ TxKey txKeyInt `elem` keys
+      _ ->
+        counterexample ("peerB action: " ++ show action) False
+  where
+    peerA, peerB :: PeerAddr
+    peerA = 1
+    peerB = 2
+
+    txid :: TxId
+    txid = 4
+
+    txKeyInt :: Int
+    txKeyInt = 0
+
+    txSize :: SizeInBytes
+    txSize = 100
+
+    txBody :: Tx TxId
+    txBody = mkTx txid txSize
+
+    sharedState0 :: SharedTxState PeerAddr TxId
+    sharedState0 =
+        ensurePeerAdvertisesTxKeys peerA [TxKey txKeyInt]
+      $ ensurePeerAdvertisesTxKeys peerB [TxKey txKeyInt]
+      $ emptySharedTxState
+          { sharedTxIdToKey = Map.singleton (getRawTxId txid) (TxKey txKeyInt)
+          , sharedKeyToTxId = IntMap.singleton txKeyInt txid
+          , sharedNextTxKey = txKeyInt + 1
+          , sharedTxTable   = IntMap.singleton txKeyInt TxEntry
+              { txLease           = TxLeased peerA (addTime 1 now)
+              , txAdvertiserCount = 2
+              , txAttempts        = Map.singleton peerA TxBuffered
+              }
+          }
+
+    peerAState :: PeerTxLocalState (Tx TxId)
+    peerAState = emptyPeerTxLocalState
+      { peerUnacknowledgedTxIds = StrictSeq.singleton (TxKey txKeyInt)
+      , peerDownloadedTxs       = IntMap.singleton txKeyInt txBody
+      }
+
+    peerBState :: PeerTxLocalState (Tx TxId)
+    peerBState = emptyPeerTxLocalState
+      { peerUnacknowledgedTxIds = StrictSeq.singleton (TxKey txKeyInt)
+      , peerAvailableTxIds      = IntMap.singleton txKeyInt txSize
+      }
+
+    (_peerAStateAfter, sharedStateAfterRejection) =
+      handleSubmittedTxs now defaultTxDecisionPolicy peerA
+        []
+        [TxKey txKeyInt]
+        peerAState
+        sharedState0
+
+    (action, _peerBStateAfter, _sharedStateFinal) =
+      nextPeerAction now defaultTxDecisionPolicy peerB peerBState
+        sharedStateAfterRejection
+
 -- Verifies that nextPeerAction can steal an expired lease for the best idle
 -- advertiser and request that tx.
 prop_nextPeerAction_claimsExpiredLease
@@ -2324,20 +2393,21 @@ prop_nextPeerAction_returnsPeerGeneration (Positive peeraddr) =
     peerState0 = emptyPeerTxLocalState { peerRequestedTxIds = maxNumTxIdsToRequest defaultTxDecisionPolicy }
     (peerAction, _, _) = nextPeerAction now defaultTxDecisionPolicy peeraddr peerState0 sharedState0
 
--- Verifies that handleSubmittedTxs bumps idle advertisers while leaving
--- submitting and waiting advertisers unchanged.
-prop_handleSubmittedTxs_bumpsIdleAdvertisers
+-- Verifies that handleSubmittedTxs bumps the generation of every other
+-- advertiser of the resolved tx, regardless of phase, while leaving the
+-- submitting peer's own generation unchanged.
+prop_handleSubmittedTxs_bumpsAdvertisers
   :: Positive Int
   -> Positive Int
   -> Positive Int
   -> TxId
   -> Positive Int
   -> Property
-prop_handleSubmittedTxs_bumpsIdleAdvertisers (Positive owner0) (Positive peerA0) (Positive peerB0) txid0 txSize0 =
+prop_handleSubmittedTxs_bumpsAdvertisers (Positive owner0) (Positive peerA0) (Positive peerB0) txid0 txSize0 =
   owner /= peerA && owner /= peerB && peerA /= peerB ==>
     conjoin
       [ sharedPeerGeneration (lookupPeerOrFail peerA sharedState') === 1
-      , sharedPeerGeneration (lookupPeerOrFail peerB sharedState') === 0
+      , sharedPeerGeneration (lookupPeerOrFail peerB sharedState') === 1
       , sharedPeerGeneration (lookupPeerOrFail owner sharedState') === 0
       ]
   where
