{-# LANGUAGE BangPatterns        #-}
{-# LANGUAGE NamedFieldPuns      #-}
{-# LANGUAGE ScopedTypeVariables #-}

module Ouroboros.Network.TxSubmission.Inbound.V2.State
  ( handleReceivedTxIds
  , handleReceivedTxs
  , handleSubmittedTxs
  , markSubmittingTxs
  , nextPeerAction
  , nextPeerActionPipelined
  , currentPeerScore
  , drainPeerScore
  , peerClaimDelay
  , applyPeerEvents
  , sweepSharedState
  ) where

import Control.Exception (assert)
import Control.Monad.Class.MonadTime.SI (DiffTime, Time, addTime, diffTime)
import Data.Foldable (toList)
import Data.IntMap.Strict qualified as IntMap
import Data.IntSet (IntSet)
import Data.IntSet qualified as IntSet
import Data.List as List (foldl')
import Data.Map.Strict qualified as Map
import Data.Sequence.Strict qualified as StrictSeq
import Data.Word (Word64)
import GHC.Stack (HasCallStack)

import Ouroboros.Network.Protocol.TxSubmission2.Type (NumTxIdsToAck,
           SizeInBytes)
import Ouroboros.Network.Tx (HasRawTxId (..))
import Ouroboros.Network.TxSubmission.Inbound.V2.Policy
import Ouroboros.Network.TxSubmission.Inbound.V2.Types

data TxIdRequestMode = AllowAnyTxIdRequests | AllowPipelinedTxIdRequests
  deriving TxIdRequestMode -> TxIdRequestMode -> Bool
(TxIdRequestMode -> TxIdRequestMode -> Bool)
-> (TxIdRequestMode -> TxIdRequestMode -> Bool)
-> Eq TxIdRequestMode
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: TxIdRequestMode -> TxIdRequestMode -> Bool
== :: TxIdRequestMode -> TxIdRequestMode -> Bool
$c/= :: TxIdRequestMode -> TxIdRequestMode -> Bool
/= :: TxIdRequestMode -> TxIdRequestMode -> Bool
Eq

-- | Precomputed context for selecting the next action for one peer.
--
data PeerActionContext peeraddr txid tx = PeerActionContext {
    -- | Current time used for lease expiry and score decay decisions.
    forall peeraddr txid tx. PeerActionContext peeraddr txid tx -> Time
pacNow          :: !Time,
    -- | Decision policy that governs request, retry, and scoring limits.
    forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> TxDecisionPolicy
pacPolicy       :: !TxDecisionPolicy,
    -- | Address of the peer whose next action is being chosen.
    forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> peeraddr
pacPeerAddr     :: !peeraddr,
    -- | Current peer-local state after local pruning has been applied.
    forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState    :: !(PeerTxLocalState tx),
    -- | This peer's contribution counters mirroring shared-state writes.
    forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight :: !PeerTxInFlight,
    -- | Shared tx-submission state after shared pruning has been applied.
    forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState  :: !(SharedTxState peeraddr txid),
    -- | Score-derived delay this peer must wait after a tx becomes claimable.
    forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> DiffTime
pacClaimDelay   :: !DiffTime
  }

-- | Decision produced by 'pickPeerActionChoice'; consumed by
-- 'applyPeerActionChoice' to compute the resulting state updates.
--
data PeerActionChoice peeraddr =
    -- | Submit the buffered bodies for the given tx keys to the mempool.
    ChooseSubmit ![TxKey]
    -- | Request tx bodies: keys to request, total expected size, and the
    -- tx table updated with the new lease assignments.
  | ChooseRequestTxs ![TxKey] !SizeInBytes !(IntMap.IntMap (TxEntry peeraddr))
    -- | Send a txid request: acknowledged keys, ack count, request count,
    -- and the updated unacknowledged queue.  The wire send site decides
    -- whether the message is blocking or pipelined based on the
    -- (post-ack) unack queue.
  | ChooseRequestTxIds ![TxKey] !NumTxIdsToAck !NumTxIdsToReq !(StrictSeq.StrictSeq TxKey)
    -- | Park until shared state advances or the wake delay elapses: the
    -- observed shared-state generation and an optional wake delay.
  | ChooseDoNothing !Word64 !(Maybe DiffTime)

-- | Build a precomputed context for selecting the next action for a peer.
--
mkPeerActionContext :: Time
                    -> TxDecisionPolicy
                    -> peeraddr
                    -> PeerTxLocalState tx
                    -> PeerTxInFlight
                    -> SharedTxState peeraddr txid
                    -> PeerActionContext peeraddr txid tx
mkPeerActionContext :: forall peeraddr tx txid.
Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> PeerActionContext peeraddr txid tx
mkPeerActionContext Time
now TxDecisionPolicy
policy peeraddr
peeraddr PeerTxLocalState tx
peerState PeerTxInFlight
peerInFlight SharedTxState peeraddr txid
sharedState =
  PeerActionContext {
    pacNow :: Time
pacNow = Time
now,
    pacPolicy :: TxDecisionPolicy
pacPolicy = TxDecisionPolicy
policy,
    pacPeerAddr :: peeraddr
pacPeerAddr = peeraddr
peeraddr,
    pacPeerState :: PeerTxLocalState tx
pacPeerState = PeerTxLocalState tx
peerState',
    pacPeerInFlight :: PeerTxInFlight
pacPeerInFlight = PeerTxInFlight
peerInFlight',
    pacSharedState :: SharedTxState peeraddr txid
pacSharedState = SharedTxState peeraddr txid
sharedState,
    pacClaimDelay :: DiffTime
pacClaimDelay = TxDecisionPolicy -> Time -> PeerScore -> DiffTime
peerClaimDelay TxDecisionPolicy
policy Time
now (PeerTxLocalState tx -> PeerScore
forall tx. PeerTxLocalState tx -> PeerScore
peerScore PeerTxLocalState tx
peerState')
    }
  where
    -- Drop locally buffered bodies whose 'sharedTxTable' entry is gone:
    -- another peer submitted the body (entry moved to 'sharedRetainedTxs',
    -- and possibly swept afterwards) or the sweep dropped the entry as an
    -- orphan after this peer's attempt was released.
    --
    -- The body can never be submitted now (no active entry to mark
    -- 'txInSubmission' on), so dropping it both frees the local buffer
    -- and unblocks 'txIdAckable', which otherwise refuses to ack while a
    -- body sits in 'peerDownloadedTxs'.
    --
    -- The same orphaned keys must also be removed from 'pifLeased' and
    -- 'pifAttempting': leaving them stale would block re-claim on
    -- re-advertisement (via 'txPeerHasAttempt') and cause the bracket
    -- finalizer to over-decrement 'txAttempt' on a freshly re-interned
    -- entry.  The other in-flight sets are unaffected: 'pifAdvertised'
    -- and 'pifAcksPending' are managed by the txid path and must stay
    -- in lockstep with the peer's queue; 'pifSubmitting' is only set
    -- via 'markSubmittingTxs', which cannot fire for an entry that's
    -- already been pruned.
    downloaded :: IntMap tx
downloaded  = PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
peerState
    downloaded' :: IntMap tx
downloaded' = IntMap tx -> IntMap (TxEntry peeraddr) -> IntMap tx
forall a b. IntMap a -> IntMap b -> IntMap a
IntMap.intersection IntMap tx
downloaded (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
sharedState)
    orphaned :: IntSet
orphaned    = IntMap tx -> IntSet
forall a. IntMap a -> IntSet
IntMap.keysSet IntMap tx
downloaded
                  IntSet -> IntSet -> IntSet
`IntSet.difference` IntMap tx -> IntSet
forall a. IntMap a -> IntSet
IntMap.keysSet IntMap tx
downloaded'

    peerState' :: PeerTxLocalState tx
peerState'
      | IntMap tx -> Bool
forall a. IntMap a -> Bool
IntMap.null IntMap tx
downloaded = PeerTxLocalState tx
peerState
      | Bool
otherwise              = PeerTxLocalState tx
peerState { peerDownloadedTxs = downloaded' }

    peerInFlight' :: PeerTxInFlight
peerInFlight'
      | IntSet -> Bool
IntSet.null IntSet
orphaned = PeerTxInFlight
peerInFlight
      | Bool
otherwise            = PeerTxInFlight
peerInFlight {
            pifLeased     = pifLeased     peerInFlight `IntSet.difference` orphaned,
            pifAttempting = pifAttempting peerInFlight `IntSet.difference` orphaned
          }

-- | Compute the next peer-local action.
nextPeerAction :: Ord peeraddr
               => Time
               -> TxDecisionPolicy
               -> peeraddr
               -> PeerTxLocalState tx
               -> PeerTxInFlight
               -> SharedTxState peeraddr txid
               -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
nextPeerAction :: forall peeraddr tx txid.
Ord peeraddr =>
Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
nextPeerAction = TxIdRequestMode
-> Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr tx txid.
Ord peeraddr =>
TxIdRequestMode
-> Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
nextPeerActionWithMode TxIdRequestMode
AllowAnyTxIdRequests
{-# INLINABLE nextPeerAction #-}

-- | Pipelined version of nextPeerAction
nextPeerActionPipelined :: Ord peeraddr
                        => Time
                        -> TxDecisionPolicy
                        -> peeraddr
                        -> PeerTxLocalState tx
                        -> PeerTxInFlight
                        -> SharedTxState peeraddr txid
                        -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
nextPeerActionPipelined :: forall peeraddr tx txid.
Ord peeraddr =>
Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
nextPeerActionPipelined = TxIdRequestMode
-> Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr tx txid.
Ord peeraddr =>
TxIdRequestMode
-> Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
nextPeerActionWithMode TxIdRequestMode
AllowPipelinedTxIdRequests
{-# INLINABLE nextPeerActionPipelined #-}

-- | V2 peer-thread scheduler
--
-- nextPeerActionWithMode handles body requests for txs this peer may currently
-- fetch, tx submission for bodies buffered locally by this peer, and txid ack/request
-- messages.  Before picking an action it runs 'bumpStuckEntries' over this
-- peer's buffered bodies to raise the inflight cap on entries whose
-- leaseholder has stalled, letting other peers claim them on their next pass.
-- Updates 'peerPhase' on the returned 'PeerTxLocalState' to reflect the
-- chosen action and threads the per-peer 'PeerTxInFlight' counters.
nextPeerActionWithMode :: Ord peeraddr
                       => TxIdRequestMode
                       -> Time
                       -> TxDecisionPolicy
                       -> peeraddr
                       -> PeerTxLocalState tx
                       -> PeerTxInFlight
                       -> SharedTxState peeraddr txid
                       -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
nextPeerActionWithMode :: forall peeraddr tx txid.
Ord peeraddr =>
TxIdRequestMode
-> Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
nextPeerActionWithMode TxIdRequestMode
txIdRequestMode Time
now TxDecisionPolicy
policy peeraddr
peeraddr PeerTxLocalState tx
peerState PeerTxInFlight
peerInFlight SharedTxState peeraddr txid
sharedState =
    let (PeerAction
action, PeerTxLocalState tx
peerState', PeerTxInFlight
peerInFlight', SharedTxState peeraddr txid
sharedState'') =
          PeerActionContext peeraddr txid tx
-> PeerActionChoice peeraddr
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> PeerActionChoice peeraddr
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyPeerActionChoice PeerActionContext peeraddr txid tx
ctx (TxIdRequestMode
-> PeerActionContext peeraddr txid tx -> PeerActionChoice peeraddr
forall peeraddr txid tx.
Ord peeraddr =>
TxIdRequestMode
-> PeerActionContext peeraddr txid tx -> PeerActionChoice peeraddr
pickPeerActionChoice TxIdRequestMode
txIdRequestMode PeerActionContext peeraddr txid tx
ctx)
        newPhase :: PeerPhase
newPhase = TxIdRequestMode -> PeerPhase -> PeerAction -> PeerPhase
phaseForAction TxIdRequestMode
txIdRequestMode (PeerTxLocalState tx -> PeerPhase
forall tx. PeerTxLocalState tx -> PeerPhase
peerPhase PeerTxLocalState tx
peerState) PeerAction
action
        peerState'' :: PeerTxLocalState tx
peerState'' = if PeerPhase
newPhase PeerPhase -> PeerPhase -> Bool
forall a. Eq a => a -> a -> Bool
== PeerTxLocalState tx -> PeerPhase
forall tx. PeerTxLocalState tx -> PeerPhase
peerPhase PeerTxLocalState tx
peerState'
                         then PeerTxLocalState tx
peerState'
                         else PeerTxLocalState tx
peerState' { peerPhase = newPhase }
    in (PeerAction
action, PeerTxLocalState tx
peerState'', PeerTxInFlight
peerInFlight', SharedTxState peeraddr txid
sharedState'')
  where
    sharedState' :: SharedTxState peeraddr txid
sharedState' = Time
-> TxDecisionPolicy
-> IntMap tx
-> SharedTxState peeraddr txid
-> SharedTxState peeraddr txid
forall peeraddr txid tx.
Time
-> TxDecisionPolicy
-> IntMap tx
-> SharedTxState peeraddr txid
-> SharedTxState peeraddr txid
bumpStuckEntries Time
now TxDecisionPolicy
policy (PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
peerState) SharedTxState peeraddr txid
sharedState
    ctx :: PeerActionContext peeraddr txid tx
ctx = Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> PeerActionContext peeraddr txid tx
forall peeraddr tx txid.
Time
-> TxDecisionPolicy
-> peeraddr
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> PeerActionContext peeraddr txid tx
mkPeerActionContext Time
now TxDecisionPolicy
policy peeraddr
peeraddr PeerTxLocalState tx
peerState PeerTxInFlight
peerInFlight SharedTxState peeraddr txid
sharedState'

-- | Compute the new 'PeerPhase' for the chosen 'PeerAction'.
--
-- In pipelined mode a 'PeerDoNothing' keeps the current phase (the peer
-- is still mid-pipeline, just waiting for replies). In non-pipelined mode
-- a 'PeerDoNothing' transitions to 'PeerIdle'.
phaseForAction :: TxIdRequestMode -> PeerPhase -> PeerAction -> PeerPhase
phaseForAction :: TxIdRequestMode -> PeerPhase -> PeerAction -> PeerPhase
phaseForAction TxIdRequestMode
txIdRequestMode PeerPhase
currentPhase PeerAction
action = case PeerAction
action of
    PeerDoNothing {}    -> case TxIdRequestMode
txIdRequestMode of
                             TxIdRequestMode
AllowPipelinedTxIdRequests -> PeerPhase
currentPhase
                             TxIdRequestMode
AllowAnyTxIdRequests       -> PeerPhase
PeerIdle
    PeerSubmitTxs {}    -> PeerPhase
PeerSubmittingToMempool
    PeerRequestTxs {}   -> PeerPhase
PeerWaitingTxs
    PeerRequestTxIds {} -> PeerPhase
PeerWaitingTxIds

-- | Pick which action to perform next.
--
pickPeerActionChoice :: Ord peeraddr
                     => TxIdRequestMode
                     -> PeerActionContext peeraddr txid tx
                     -> PeerActionChoice peeraddr
pickPeerActionChoice :: forall peeraddr txid tx.
Ord peeraddr =>
TxIdRequestMode
-> PeerActionContext peeraddr txid tx -> PeerActionChoice peeraddr
pickPeerActionChoice TxIdRequestMode
txIdRequestMode PeerActionContext peeraddr txid tx
ctx
  -- Pick TXs to submit to the mempool
  | Just [TxKey]
txsToSubmit <- PeerActionContext peeraddr txid tx -> Maybe [TxKey]
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> Maybe [TxKey]
pickSubmitAction PeerActionContext peeraddr txid tx
ctx =
      [TxKey] -> PeerActionChoice peeraddr
forall peeraddr. [TxKey] -> PeerActionChoice peeraddr
ChooseSubmit [TxKey]
txsToSubmit
  -- Pick TXs to fetch
  | Just ([TxKey]
txsToRequest, SizeInBytes
txsToRequestSize, IntMap (TxEntry peeraddr)
txTable') <- PeerActionContext peeraddr txid tx
-> Maybe ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
forall peeraddr txid tx.
Eq peeraddr =>
PeerActionContext peeraddr txid tx
-> Maybe ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
pickRequestTxsAction PeerActionContext peeraddr txid tx
ctx =
      [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> PeerActionChoice peeraddr
forall peeraddr.
[TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> PeerActionChoice peeraddr
ChooseRequestTxs [TxKey]
txsToRequest SizeInBytes
txsToRequestSize IntMap (TxEntry peeraddr)
txTable'
  -- Pick TXids to ack and/or request more TXids.
  | Just ([TxKey]
acknowledgedTxIds, NumTxIdsToAck
txIdsToAcknowledge, NumTxIdsToReq
txIdsToRequest, StrictSeq TxKey
unacknowledgedTxIds') <-
      TxIdRequestMode
-> PeerActionContext peeraddr txid tx
-> Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall peeraddr txid tx.
TxIdRequestMode
-> PeerActionContext peeraddr txid tx
-> Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
pickRequestTxIdsAction TxIdRequestMode
txIdRequestMode PeerActionContext peeraddr txid tx
ctx =
      [TxKey]
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StrictSeq TxKey
-> PeerActionChoice peeraddr
forall peeraddr.
[TxKey]
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StrictSeq TxKey
-> PeerActionChoice peeraddr
ChooseRequestTxIds [TxKey]
acknowledgedTxIds NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToReq
txIdsToRequest StrictSeq TxKey
unacknowledgedTxIds'
  -- Do nothing
  | Bool
otherwise =
      Word64 -> Maybe DiffTime -> PeerActionChoice peeraddr
forall peeraddr.
Word64 -> Maybe DiffTime -> PeerActionChoice peeraddr
ChooseDoNothing (SharedTxState peeraddr txid -> Word64
forall peeraddr txid. SharedTxState peeraddr txid -> Word64
sharedGeneration (PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState PeerActionContext peeraddr txid tx
ctx)) (PeerActionContext peeraddr txid tx -> Maybe DiffTime
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> Maybe DiffTime
nextWakeDelay PeerActionContext peeraddr txid tx
ctx)

-- | Execute a chosen peer action and compute resulting state updates
applyPeerActionChoice :: PeerActionContext peeraddr txid tx
                      -> PeerActionChoice peeraddr
                      -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
applyPeerActionChoice :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> PeerActionChoice peeraddr
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyPeerActionChoice PeerActionContext peeraddr txid tx
ctx PeerActionChoice peeraddr
choice =
  case PeerActionChoice peeraddr
choice of
       ChooseSubmit [TxKey]
txsToSubmit ->
         PeerActionContext peeraddr txid tx
-> [TxKey]
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> [TxKey]
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applySubmitChoice PeerActionContext peeraddr txid tx
ctx [TxKey]
txsToSubmit
       ChooseRequestTxs [TxKey]
txsToRequest SizeInBytes
txsToRequestSize IntMap (TxEntry peeraddr)
txTable' ->
         PeerActionContext peeraddr txid tx
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyRequestTxsChoice PeerActionContext peeraddr txid tx
ctx [TxKey]
txsToRequest SizeInBytes
txsToRequestSize IntMap (TxEntry peeraddr)
txTable'
       ChooseRequestTxIds [TxKey]
acknowledgedTxIds NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToReq
txIdsToRequest
                          StrictSeq TxKey
unacknowledgedTxIds' ->
         PeerActionContext peeraddr txid tx
-> [TxKey]
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StrictSeq TxKey
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> [TxKey]
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StrictSeq TxKey
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyRequestTxIdsChoice PeerActionContext peeraddr txid tx
ctx [TxKey]
acknowledgedTxIds NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToReq
txIdsToRequest
                                 StrictSeq TxKey
unacknowledgedTxIds'
       ChooseDoNothing Word64
generation Maybe DiffTime
wakeDelay ->
         PeerActionContext peeraddr txid tx
-> Word64
-> Maybe DiffTime
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> Word64
-> Maybe DiffTime
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyDoNothingChoice PeerActionContext peeraddr txid tx
ctx Word64
generation Maybe DiffTime
wakeDelay

-- | Construct a 'PeerSubmitTxs' action for buffered transactions.
--
-- Marks the selected txs as in-submission on this peer.  Other peers
-- skip them while the 'txInSubmission' flag is set.  STM serialisation
-- guarantees only one peer can win the @ChooseSubmit@ race for a given key.
applySubmitChoice :: PeerActionContext peeraddr txid tx
                  -> [TxKey]
                  -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
applySubmitChoice :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> [TxKey]
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applySubmitChoice PeerActionContext peeraddr txid tx
ctx [TxKey]
txsToSubmit =
  let keys :: IntSet
keys = [Key] -> IntSet
IntSet.fromList (TxKey -> Key
unTxKey (TxKey -> Key) -> [TxKey] -> [Key]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [TxKey]
txsToSubmit)
      pif :: PeerTxInFlight
pif  = PeerActionContext peeraddr txid tx -> PeerTxInFlight
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight PeerActionContext peeraddr txid tx
ctx
      pif' :: PeerTxInFlight
pif' = PeerTxInFlight
pif {
          pifAttempting = pifAttempting pif `IntSet.difference` keys,
          pifSubmitting = pifSubmitting pif `IntSet.union`      keys
        }
  in ( [TxKey] -> PeerAction
PeerSubmitTxs [TxKey]
txsToSubmit
     , PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState PeerActionContext peeraddr txid tx
ctx
     , PeerTxInFlight
pif'
     , [TxKey]
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
forall peeraddr txid.
[TxKey]
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
markSubmittingTxs [TxKey]
txsToSubmit (PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState PeerActionContext peeraddr txid tx
ctx)
     )

-- | Construct a 'PeerRequestTxs' action and update local and shared tx state.
applyRequestTxsChoice :: PeerActionContext peeraddr txid tx
                      -> [TxKey]
                      -> SizeInBytes
                      -> IntMap.IntMap (TxEntry peeraddr)
                      -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
applyRequestTxsChoice :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyRequestTxsChoice PeerActionContext peeraddr txid tx
ctx [TxKey]
txsToRequest SizeInBytes
txsToRequestSize IntMap (TxEntry peeraddr)
txTable =
  ( [TxKey] -> PeerAction
PeerRequestTxs [TxKey]
txsToRequest
  , PeerTxLocalState tx
peerState'
  , PeerTxInFlight
peerInFlight'
  , SharedTxState peeraddr txid
sharedState'
  )
  where
    requestedKeys :: IntSet
requestedKeys = [Key] -> IntSet
IntSet.fromList (TxKey -> Key
unTxKey (TxKey -> Key) -> [TxKey] -> [Key]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [TxKey]
txsToRequest)
    peerState' :: PeerTxLocalState tx
peerState' =
      (PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState PeerActionContext peeraddr txid tx
ctx) {
        peerRequestedTxs =
          peerRequestedTxs (pacPeerState ctx) `IntSet.union` requestedKeys,
        peerRequestedTxBatches =
          peerRequestedTxBatches (pacPeerState ctx) StrictSeq.|> RequestedTxBatch {
            requestedTxBatchSet = requestedKeys,
            requestedTxBatchSize = txsToRequestSize
          },
        peerRequestedTxsSize = peerRequestedTxsSize (pacPeerState ctx) + txsToRequestSize
      }
    pif :: PeerTxInFlight
pif = PeerActionContext peeraddr txid tx -> PeerTxInFlight
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight PeerActionContext peeraddr txid tx
ctx
    peerInFlight' :: PeerTxInFlight
peerInFlight' = PeerTxInFlight
pif {
        pifLeased     = pifLeased     pif `IntSet.union` requestedKeys,
        pifAttempting = pifAttempting pif `IntSet.union` requestedKeys
      }
    -- Bump only 'sharedRevision' (the structural dirty bit), not
    -- 'sharedGeneration' (the wake counter).  Claiming a lease grants no
    -- new option to other advertisers: they couldn't claim before this
    -- commit, and they still can't.  Waking them here would cost a full
    -- 'nextPeerAction' pass per peer with nothing to show for it; the
    -- wake they actually care about is on submit / lease release.
    sharedState' :: SharedTxState peeraddr txid
sharedState' =
      (PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState PeerActionContext peeraddr txid tx
ctx) {
        sharedTxTable = txTable,
        sharedRevision = sharedRevision (pacSharedState ctx) + 1
      }

-- | Construct a 'PeerRequestTxIds' action and update local and shared txid state.
applyRequestTxIdsChoice
  :: PeerActionContext peeraddr txid tx
  -> [TxKey]
  -> NumTxIdsToAck
  -> NumTxIdsToReq
  -> StrictSeq.StrictSeq TxKey
  -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
applyRequestTxIdsChoice :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> [TxKey]
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StrictSeq TxKey
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyRequestTxIdsChoice PeerActionContext peeraddr txid tx
ctx [TxKey]
acknowledgedTxIds NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToReq
txIdsToRequest StrictSeq TxKey
unacknowledgedTxIds' =
  ( NumTxIdsToAck -> NumTxIdsToReq -> PeerAction
PeerRequestTxIds NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToReq
txIdsToRequest
  , PeerTxLocalState tx
peerState''
  , PeerTxInFlight
peerInFlight''
  , PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState PeerActionContext peeraddr txid tx
ctx
  )
  where
    peerState0 :: PeerTxLocalState tx
peerState0 = PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState PeerActionContext peeraddr txid tx
ctx
    acknowledgedKeys :: IntSet
acknowledgedKeys = [Key] -> IntSet
IntSet.fromList (TxKey -> Key
unTxKey (TxKey -> Key) -> [TxKey] -> [Key]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [TxKey]
acknowledgedTxIds)
    peerState'' :: PeerTxLocalState tx
peerState'' =
      PeerTxLocalState tx
peerState0 {
        peerAvailableTxIds =
          IntMap.withoutKeys (peerAvailableTxIds peerState0) acknowledgedKeys,
        peerUnacknowledgedTxIds = unacknowledgedTxIds',
        peerRequestedTxIds = peerRequestedTxIds peerState0 + txIdsToRequest
      }
    pif :: PeerTxInFlight
pif = PeerActionContext peeraddr txid tx -> PeerTxInFlight
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight PeerActionContext peeraddr txid tx
ctx
    peerInFlight'' :: PeerTxInFlight
peerInFlight'' = PeerTxInFlight
pif {
        pifAdvertised  = pifAdvertised  pif `IntSet.difference` acknowledgedKeys,
        pifAcksPending = pifAcksPending pif `IntSet.difference` acknowledgedKeys
      }

-- | Construct a 'PeerDoNothing' action.
applyDoNothingChoice
  :: PeerActionContext peeraddr txid tx
  -> Word64
  -> Maybe DiffTime
  -> (PeerAction, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
applyDoNothingChoice :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx
-> Word64
-> Maybe DiffTime
-> (PeerAction, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
applyDoNothingChoice PeerActionContext peeraddr txid tx
ctx Word64
generation Maybe DiffTime
wakeDelay =
  ( Word64 -> Maybe DiffTime -> PeerAction
PeerDoNothing Word64
generation Maybe DiffTime
wakeDelay
  , PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState PeerActionContext peeraddr txid tx
ctx
  , PeerActionContext peeraddr txid tx -> PeerTxInFlight
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight PeerActionContext peeraddr txid tx
ctx
  , PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState PeerActionContext peeraddr txid tx
ctx
  )

-- | Select downloaded transactions that this peer may submit to the mempool.
pickSubmitAction
  :: PeerActionContext peeraddr txid tx
  -> Maybe [TxKey]
pickSubmitAction :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> Maybe [TxKey]
pickSubmitAction PeerActionContext { PeerTxLocalState tx
pacPeerState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState :: PeerTxLocalState tx
pacPeerState, SharedTxState peeraddr txid
pacSharedState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState :: SharedTxState peeraddr txid
pacSharedState } =
  if [TxKey] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [TxKey]
pickBufferedTxsToSubmit
     then Maybe [TxKey]
forall a. Maybe a
Nothing
     else [TxKey] -> Maybe [TxKey]
forall a. a -> Maybe a
Just [TxKey]
pickBufferedTxsToSubmit
  where

    -- Walk the unacknowledged txid queue in peer advertisement order,
    -- picking bodies buffered by this peer for immediate submission.
    -- Classification of each entry:
    --
    -- * 'sharedTxTable' has 'Nothing' for the key: the tx was resolved at
    --   some point (a peer submitted it to the mempool) and the entry was
    --   deleted.  Orphan sweep cannot drop an entry while any peer still
    --   tracks the key, so 'Nothing' here implies resolution; safe to skip
    --   and continue past, regardless of whether the retained marker has
    --   since expired or the mempool has since evicted.
    -- * 'sharedTxTable' has 'Just' but we don't have the body buffered, or
    --   another peer is already submitting: the tx is in flight elsewhere.
    --   Stop, otherwise later txs in our stream might run ahead of an
    --   unresolved earlier tx they depend on.
    -- * 'sharedTxTable' has 'Just' and we have the body buffered: submit.
    --
    -- The same 'TxKey' may legitimately appear more than once in
    -- 'peerUnacknowledgedTxIds' (e.g. a fork switch on the sender re-adds
    -- a tx to the mempool under a new idx, see 'txSubmissionOutbound').
    -- Track already-picked keys in 'seen' so each key is submitted at
    -- most once per scheduler invocation; the duplicates remain in the
    -- queue so the ack-by-count handshake stays in sync with the sender.
    pickBufferedTxsToSubmit :: [TxKey]
pickBufferedTxsToSubmit =
        IntSet -> [TxKey] -> [TxKey] -> [TxKey]
go IntSet
IntSet.empty [] (StrictSeq TxKey -> [TxKey]
forall a. StrictSeq a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState))
      where
        go :: IntSet -> [TxKey] -> [TxKey] -> [TxKey]
go IntSet
_    [TxKey]
acc [] = [TxKey] -> [TxKey]
forall a. [a] -> [a]
reverse [TxKey]
acc
        go IntSet
seen [TxKey]
acc (txKey :: TxKey
txKey@(TxKey Key
k) : [TxKey]
rest)
          | Key -> IntSet -> Bool
IntSet.member Key
k IntSet
seen = IntSet -> [TxKey] -> [TxKey] -> [TxKey]
go IntSet
seen [TxKey]
acc [TxKey]
rest
          | Bool
otherwise =
              case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
pacSharedState) of
                Just TxEntry peeraddr
txEntry
                  | PeerTxLocalState tx -> Key -> Bool
forall tx. PeerTxLocalState tx -> Key -> Bool
txBufferedByPeer PeerTxLocalState tx
pacPeerState Key
k
                  , Bool -> Bool
not (TxEntry peeraddr -> Bool
forall peeraddr. TxEntry peeraddr -> Bool
txInSubmission TxEntry peeraddr
txEntry) ->
                      IntSet -> [TxKey] -> [TxKey] -> [TxKey]
go (Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
seen) (TxKey
txKey TxKey -> [TxKey] -> [TxKey]
forall a. a -> [a] -> [a]
: [TxKey]
acc) [TxKey]
rest
                Just TxEntry peeraddr
_  -> [TxKey] -> [TxKey]
forall a. [a] -> [a]
reverse [TxKey]
acc
                Maybe (TxEntry peeraddr)
Nothing -> IntSet -> [TxKey] -> [TxKey] -> [TxKey]
go IntSet
seen [TxKey]
acc [TxKey]
rest

-- | Select transactions to request from the peer, if within policy limits.
--
-- Returns a triple of:
-- Tx keys to request (in the peer's advertisement order)
-- Total serialized size of the requested txs
-- Updated shared state with new lease ownership for selected txs
pickRequestTxsAction :: Eq peeraddr
                     => PeerActionContext peeraddr txid tx
                     -> Maybe ([TxKey], SizeInBytes, IntMap.IntMap (TxEntry peeraddr))
pickRequestTxsAction :: forall peeraddr txid tx.
Eq peeraddr =>
PeerActionContext peeraddr txid tx
-> Maybe ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
pickRequestTxsAction ctx :: PeerActionContext peeraddr txid tx
ctx@PeerActionContext { Time
pacNow :: forall peeraddr txid tx. PeerActionContext peeraddr txid tx -> Time
pacNow :: Time
pacNow, TxDecisionPolicy
pacPolicy :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> TxDecisionPolicy
pacPolicy :: TxDecisionPolicy
pacPolicy, PeerTxLocalState tx
pacPeerState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState :: PeerTxLocalState tx
pacPeerState, SharedTxState peeraddr txid
pacSharedState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState :: SharedTxState peeraddr txid
pacSharedState } =
  let ([TxKey]
txsToRequest, SizeInBytes
txsToRequestSize, IntMap (TxEntry peeraddr)
sharedState') = ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
pickTxsToRequest in
  if [TxKey] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [TxKey]
txsToRequest
     then Maybe ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
forall a. Maybe a
Nothing
     else ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
-> Maybe ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
forall a. a -> Maybe a
Just ([TxKey]
txsToRequest, SizeInBytes
txsToRequestSize, IntMap (TxEntry peeraddr)
sharedState')
  where

    -- Picks txs from the peer's available set that are not yet requested or
    -- downloaded, assigning leases with expiry timestamps.
    -- Respects 'maxOutstandingTxBatchesPerPeer' and 'txsSizeInflightPerPeer' policy
    -- constraints.
    pickTxsToRequest :: ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
pickTxsToRequest =
      if StrictSeq RequestedTxBatch -> Key
forall a. StrictSeq a -> Key
StrictSeq.length (PeerTxLocalState tx -> StrictSeq RequestedTxBatch
forall tx. PeerTxLocalState tx -> StrictSeq RequestedTxBatch
peerRequestedTxBatches PeerTxLocalState tx
pacPeerState) Key -> Key -> Bool
forall a. Ord a => a -> a -> Bool
>=
           TxDecisionPolicy -> Key
maxOutstandingTxBatchesPerPeer TxDecisionPolicy
pacPolicy
         then ([], SizeInBytes
0, SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
pacSharedState)
         else IntSet
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> [(Key, SizeInBytes)]
-> ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
go IntSet
IntSet.empty [] SizeInBytes
0 (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
pacSharedState) [(Key, SizeInBytes)]
candidates
      where
        -- Remaining bytes available for requesting new tx, based on the
        -- per-peer inflight size limit.
        sizeBudget :: SizeInBytes
sizeBudget =
          if PeerTxLocalState tx -> SizeInBytes
forall tx. PeerTxLocalState tx -> SizeInBytes
peerRequestedTxsSize PeerTxLocalState tx
pacPeerState SizeInBytes -> SizeInBytes -> Bool
forall a. Ord a => a -> a -> Bool
>= TxDecisionPolicy -> SizeInBytes
txsSizeInflightPerPeer TxDecisionPolicy
pacPolicy
             then SizeInBytes
0
             else TxDecisionPolicy -> SizeInBytes
txsSizeInflightPerPeer TxDecisionPolicy
pacPolicy SizeInBytes -> SizeInBytes -> SizeInBytes
forall a. Num a => a -> a -> a
- PeerTxLocalState tx -> SizeInBytes
forall tx. PeerTxLocalState tx -> SizeInBytes
peerRequestedTxsSize PeerTxLocalState tx
pacPeerState

        leaseUntil :: Time
leaseUntil = DiffTime -> Time -> Time
addTime (TxDecisionPolicy -> DiffTime
interTxSpace TxDecisionPolicy
pacPolicy) Time
pacNow

        -- Iterate the peer's unacknowledged queue, which preserves the order
        -- in which the peer sent us the txids.  Walking the queue in that
        -- order aligns our fetch order with the peer's advertisement order,
        -- which preserves whatever submission-validity ordering the peer
        -- intended within the batch.
        candidates :: [(Key, SizeInBytes)]
candidates =
          [ (Key
k, SizeInBytes
txSize)
          | TxKey Key
k <- StrictSeq TxKey -> [TxKey]
forall a. StrictSeq a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState)
          , Key -> IntSet -> Bool
IntSet.notMember Key
k (PeerTxLocalState tx -> IntSet
forall tx. PeerTxLocalState tx -> IntSet
peerRequestedTxs PeerTxLocalState tx
pacPeerState)
          , Key -> IntMap tx -> Bool
forall a. Key -> IntMap a -> Bool
IntMap.notMember Key
k (PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
pacPeerState)
          , Just SizeInBytes
txSize <- [Key -> IntMap SizeInBytes -> Maybe SizeInBytes
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (PeerTxLocalState tx -> IntMap SizeInBytes
forall tx. PeerTxLocalState tx -> IntMap SizeInBytes
peerAvailableTxIds PeerTxLocalState tx
pacPeerState)]
          ]

        -- Select transactions to request by iterating through candidates in
        -- peer advertisement order until the size budget is consumed.
        --
        -- The same key may legitimately appear more than once in the
        -- queue (see 'pickBufferedTxsToSubmit' for the fork-switch
        -- scenario).  Track keys already picked this round in 'seen' so
        -- 'claimTx' fires at most once per key; the wire-level request
        -- 'Map txid SizeInBytes' already deduplicates, and we must not
        -- inflate 'txAttempt' or 'peerRequestedTxsSize' with phantom
        -- second claims.
        go :: IntSet
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> [(Key, SizeInBytes)]
-> ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
go IntSet
_    [TxKey]
selectedRev SizeInBytes
selectedSize IntMap (TxEntry peeraddr)
txTable [] = ([TxKey] -> [TxKey]
forall a. [a] -> [a]
reverse [TxKey]
selectedRev, SizeInBytes
selectedSize, IntMap (TxEntry peeraddr)
txTable)
        go IntSet
seen [TxKey]
selectedRev SizeInBytes
selectedSize IntMap (TxEntry peeraddr)
txTable ((Key
k, SizeInBytes
txSize) : [(Key, SizeInBytes)]
rest)
          | Key -> IntSet -> Bool
IntSet.member Key
k IntSet
seen = IntSet
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> [(Key, SizeInBytes)]
-> ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
go IntSet
seen [TxKey]
selectedRev SizeInBytes
selectedSize IntMap (TxEntry peeraddr)
txTable [(Key, SizeInBytes)]
rest
          | SizeInBytes -> SizeInBytes -> Bool
exceedsBudget SizeInBytes
selectedSize SizeInBytes
txSize =
              ([TxKey] -> [TxKey]
forall a. [a] -> [a]
reverse [TxKey]
selectedRev, SizeInBytes
selectedSize, IntMap (TxEntry peeraddr)
txTable)
          | Bool
otherwise =
              case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k IntMap (TxEntry peeraddr)
txTable of
                  Just TxEntry peeraddr
txEntry ->
                    if PeerActionContext peeraddr txid tx
-> TxKey -> TxEntry peeraddr -> Bool
forall peeraddr txid tx.
Eq peeraddr =>
PeerActionContext peeraddr txid tx
-> TxKey -> TxEntry peeraddr -> Bool
txSelectable PeerActionContext peeraddr txid tx
ctx (Key -> TxKey
TxKey Key
k) TxEntry peeraddr
txEntry
                       then
                         IntSet
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> [(Key, SizeInBytes)]
-> ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
go (Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
seen)
                            (Key -> TxKey
TxKey Key
k TxKey -> [TxKey] -> [TxKey]
forall a. a -> [a] -> [a]
: [TxKey]
selectedRev)
                            (SizeInBytes
selectedSize SizeInBytes -> SizeInBytes -> SizeInBytes
forall a. Num a => a -> a -> a
+ SizeInBytes
txSize)
                            (Key
-> TxEntry peeraddr
-> IntMap (TxEntry peeraddr)
-> IntMap (TxEntry peeraddr)
forall a. Key -> a -> IntMap a -> IntMap a
IntMap.insert Key
k (peeraddr -> Time -> TxEntry peeraddr -> TxEntry peeraddr
forall peeraddr.
peeraddr -> Time -> TxEntry peeraddr -> TxEntry peeraddr
claimTx (PeerActionContext peeraddr txid tx -> peeraddr
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> peeraddr
pacPeerAddr PeerActionContext peeraddr txid tx
ctx) Time
leaseUntil TxEntry peeraddr
txEntry)
                                           IntMap (TxEntry peeraddr)
txTable)
                            [(Key, SizeInBytes)]
rest
                       else IntSet
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> [(Key, SizeInBytes)]
-> ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
go IntSet
seen [TxKey]
selectedRev SizeInBytes
selectedSize IntMap (TxEntry peeraddr)
txTable [(Key, SizeInBytes)]
rest
                  -- Entry resolved or swept since this peer's local state last synced; skip.
                  Maybe (TxEntry peeraddr)
Nothing -> IntSet
-> [TxKey]
-> SizeInBytes
-> IntMap (TxEntry peeraddr)
-> [(Key, SizeInBytes)]
-> ([TxKey], SizeInBytes, IntMap (TxEntry peeraddr))
go IntSet
seen [TxKey]
selectedRev SizeInBytes
selectedSize IntMap (TxEntry peeraddr)
txTable [(Key, SizeInBytes)]
rest

        -- Check if the size is within the budget.
        --
        -- The inflight size limit is soft by up to one tx size, so a peer with
        -- spare capacity may still request its first tx in a batch even when
        -- that single tx exceeds the remaining byte budget.
        exceedsBudget :: SizeInBytes -> SizeInBytes -> Bool
exceedsBudget SizeInBytes
selectedSize SizeInBytes
txSize
          | SizeInBytes
selectedSize SizeInBytes -> SizeInBytes -> SizeInBytes
forall a. Num a => a -> a -> a
+ SizeInBytes
txSize SizeInBytes -> SizeInBytes -> Bool
forall a. Ord a => a -> a -> Bool
<= SizeInBytes
sizeBudget = Bool
False
          | SizeInBytes
selectedSize SizeInBytes -> SizeInBytes -> Bool
forall a. Eq a => a -> a -> Bool
/= SizeInBytes
0 = Bool
True
          | Bool
otherwise = PeerTxLocalState tx -> SizeInBytes
forall tx. PeerTxLocalState tx -> SizeInBytes
peerRequestedTxsSize PeerTxLocalState tx
pacPeerState SizeInBytes -> SizeInBytes -> Bool
forall a. Ord a => a -> a -> Bool
>= TxDecisionPolicy -> SizeInBytes
txsSizeInflightPerPeer TxDecisionPolicy
pacPolicy

-- | Determine txid acknowledgment and request counts, if any work is available.
pickRequestTxIdsAction :: TxIdRequestMode
                       -> PeerActionContext peeraddr txid tx
                       -> Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq.StrictSeq TxKey)
pickRequestTxIdsAction :: forall peeraddr txid tx.
TxIdRequestMode
-> PeerActionContext peeraddr txid tx
-> Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
pickRequestTxIdsAction TxIdRequestMode
txIdRequestMode ctx :: PeerActionContext peeraddr txid tx
ctx@PeerActionContext { TxDecisionPolicy
pacPolicy :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> TxDecisionPolicy
pacPolicy :: TxDecisionPolicy
pacPolicy, PeerTxLocalState tx
pacPeerState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState :: PeerTxLocalState tx
pacPeerState }
  | NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToAck -> NumTxIdsToAck -> Bool
forall a. Ord a => a -> a -> Bool
<= NumTxIdsToAck
0 Bool -> Bool -> Bool
&& NumTxIdsToReq
txIdsToRequest NumTxIdsToReq -> NumTxIdsToReq -> Bool
forall a. Ord a => a -> a -> Bool
<= NumTxIdsToReq
0 = Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall a. Maybe a
Nothing

  -- Benchmark hook: when 'disablePipelinedTxIdRequests' is set, never
  -- fire a request that would be sent as pipelined.  Pipelined fires
  -- whenever the post-ack queue is non-empty (or we're already in
  -- pipelined mode), so this guard suppresses both.  The peer then
  -- parks until the queue can drain via acks, yielding pure blocking
  -- request behaviour at the wire.
  | TxDecisionPolicy -> Bool
disablePipelinedTxIdRequests TxDecisionPolicy
pacPolicy
  , TxIdRequestMode
txIdRequestMode TxIdRequestMode -> TxIdRequestMode -> Bool
forall a. Eq a => a -> a -> Bool
== TxIdRequestMode
AllowPipelinedTxIdRequests
    Bool -> Bool -> Bool
|| Bool -> Bool
not (StrictSeq TxKey -> Bool
forall a. StrictSeq a -> Bool
StrictSeq.null StrictSeq TxKey
unacknowledgedTxIds) = Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall a. Maybe a
Nothing

  -- A pure-ack pipelined message would burn a pipeline slot for an
  -- empty reply ('req=0' forces the response empty by construction) and
  -- shrink our window without growing it.  Defer the ack until we can
  -- also request more txids.
  | TxIdRequestMode
txIdRequestMode TxIdRequestMode -> TxIdRequestMode -> Bool
forall a. Eq a => a -> a -> Bool
== TxIdRequestMode
AllowPipelinedTxIdRequests
  , NumTxIdsToReq
txIdsToRequest NumTxIdsToReq -> NumTxIdsToReq -> Bool
forall a. Ord a => a -> a -> Bool
<= NumTxIdsToReq
0 = Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall a. Maybe a
Nothing

  -- Spec: a pipelined (non-blocking) request requires the post-ack
  -- queue to be non-empty.  When the peer's unacked queue is empty to
  -- begin with we can't satisfy that, and there's no blocking option
  -- in pipelined mode, so wait for a later wake.
  | TxIdRequestMode
txIdRequestMode TxIdRequestMode -> TxIdRequestMode -> Bool
forall a. Eq a => a -> a -> Bool
== TxIdRequestMode
AllowPipelinedTxIdRequests
  , StrictSeq TxKey -> Bool
forall a. StrictSeq a -> Bool
StrictSeq.null (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState) = Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall a. Maybe a
Nothing

  -- Backoff after an empty txid reply: if our last request came back
  -- with zero txids and we still have nothing to ack, suppress the
  -- next 'ack=0, req=N' pipelined wire message (V2.hs sends pipelined
  -- whenever the unacked queue is non-empty).  The flag is cleared by
  -- a non-empty txid reply or by the mempool accepting one of this
  -- peer's txs - both signs that the peer may have new work for us.
  | PeerTxLocalState tx -> Bool
forall tx. PeerTxLocalState tx -> Bool
peerLastTxIdReplyWasEmpty PeerTxLocalState tx
pacPeerState
  , NumTxIdsToAck
txIdsToAcknowledge NumTxIdsToAck -> NumTxIdsToAck -> Bool
forall a. Ord a => a -> a -> Bool
<= NumTxIdsToAck
0
  , Bool -> Bool
not (StrictSeq TxKey -> Bool
forall a. StrictSeq a -> Bool
StrictSeq.null (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState)) = Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall a. Maybe a
Nothing

  | Bool
otherwise = ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
-> Maybe ([TxKey], NumTxIdsToAck, NumTxIdsToReq, StrictSeq TxKey)
forall a. a -> Maybe a
Just ([TxKey]
acknowledgedTxIds, NumTxIdsToAck
txIdsToAcknowledge, NumTxIdsToReq
txIdsToRequest, StrictSeq TxKey
unacknowledgedTxIds)
  where

    -- Split the unacknowledged txid queue into acknowledged and remaining portions.
    --
    -- acknowledgedTxIds is the longest prefix of "ackable" txids.
    -- unacknowledgedTxIds is the remaining txids
    acknowledgedTxIds :: [TxKey]
acknowledgedTxIds = StrictSeq TxKey -> [TxKey]
forall a. StrictSeq a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList StrictSeq TxKey
acknowledgedTxIdsSeq
    txIdsToAcknowledge :: NumTxIdsToAck
txIdsToAcknowledge = Key -> NumTxIdsToAck
forall a b. (Integral a, Num b) => a -> b
fromIntegral Key
numOfAcked

    ackablePrefix :: StrictSeq TxKey
ackablePrefix =
      (TxKey -> Bool) -> StrictSeq TxKey -> StrictSeq TxKey
forall a. (a -> Bool) -> StrictSeq a -> StrictSeq a
StrictSeq.takeWhileL (PeerActionContext peeraddr txid tx -> TxKey -> Bool
forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> TxKey -> Bool
txIdAckable PeerActionContext peeraddr txid tx
ctx)
                           (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState)

    numOfUnacked :: Key
numOfUnacked = StrictSeq TxKey -> Key
forall a. StrictSeq a -> Key
StrictSeq.length (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState)
    numOfRequested :: Key
numOfRequested = NumTxIdsToReq -> Key
forall a b. (Integral a, Num b) => a -> b
fromIntegral (PeerTxLocalState tx -> NumTxIdsToReq
forall tx. PeerTxLocalState tx -> NumTxIdsToReq
peerRequestedTxIds PeerTxLocalState tx
pacPeerState) :: Int
    hasOutstandingBodyReplies :: Bool
hasOutstandingBodyReplies =
      Bool -> Bool
not (StrictSeq RequestedTxBatch -> Bool
forall a. StrictSeq a -> Bool
StrictSeq.null (PeerTxLocalState tx -> StrictSeq RequestedTxBatch
forall tx. PeerTxLocalState tx -> StrictSeq RequestedTxBatch
peerRequestedTxBatches PeerTxLocalState tx
pacPeerState))
    keepOneUnackedForPipelinedRequest :: Bool
keepOneUnackedForPipelinedRequest =
      TxIdRequestMode
txIdRequestMode TxIdRequestMode -> TxIdRequestMode -> Bool
forall a. Eq a => a -> a -> Bool
== TxIdRequestMode
AllowPipelinedTxIdRequests
        Bool -> Bool -> Bool
&& (Key
numOfRequested Key -> Key -> Bool
forall a. Ord a => a -> a -> Bool
> Key
0 Bool -> Bool -> Bool
|| Bool
hasOutstandingBodyReplies)
    numOfAcked0 :: Key
numOfAcked0 = StrictSeq TxKey -> Key
forall a. StrictSeq a -> Key
StrictSeq.length StrictSeq TxKey
ackablePrefix
    numOfAcked :: Key
numOfAcked
      -- A pipelined txid request becomes a non-blocking protocol message
      -- while any txid or body reply is still in flight. The outbound side
      -- requires at least one txid to remain unacknowledged in that case.
      | Bool
keepOneUnackedForPipelinedRequest =
          Key -> Key -> Key
forall a. Ord a => a -> a -> a
min Key
numOfAcked0 (Key -> Key -> Key
forall a. Ord a => a -> a -> a
max Key
0 (Key
numOfUnacked Key -> Key -> Key
forall a. Num a => a -> a -> a
- Key
1))
      | Bool
otherwise = Key
numOfAcked0

    (StrictSeq TxKey
acknowledgedTxIdsSeq, StrictSeq TxKey
unacknowledgedTxIds) = Key -> StrictSeq TxKey -> (StrictSeq TxKey, StrictSeq TxKey)
forall a. Key -> StrictSeq a -> (StrictSeq a, StrictSeq a)
StrictSeq.splitAt Key
numOfAcked
      (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
pacPeerState)
    unackedAndRequested :: Key
unackedAndRequested = Key
numOfUnacked Key -> Key -> Key
forall a. Num a => a -> a -> a
+ Key
numOfRequested

    -- How many new txids we can request: capped by the unack-window
    -- room left after this round's ack ('maxUnacknowledgedTxIds -
    -- unackedAndRequested + numOfAcked') and by the per-message
    -- request limit ('maxNumTxIdsToRequest - numOfRequested').
    txIdsToRequest :: NumTxIdsToReq
txIdsToRequest =
      Key -> NumTxIdsToReq
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Key -> NumTxIdsToReq) -> Key -> NumTxIdsToReq
forall a b. (a -> b) -> a -> b
$ Key -> Key -> Key
forall a. Ord a => a -> a -> a
max Key
0 (Key -> Key) -> Key -> Key
forall a b. (a -> b) -> a -> b
$ Key -> Key -> Key
forall a. Ord a => a -> a -> a
min
        (NumTxIdsToReq -> Key
forall a b. (Integral a, Num b) => a -> b
fromIntegral (TxDecisionPolicy -> NumTxIdsToReq
maxUnacknowledgedTxIds TxDecisionPolicy
pacPolicy) Key -> Key -> Key
forall a. Num a => a -> a -> a
- Key
unackedAndRequested Key -> Key -> Key
forall a. Num a => a -> a -> a
+ Key
numOfAcked)
        (NumTxIdsToReq -> Key
forall a b. (Integral a, Num b) => a -> b
fromIntegral (TxDecisionPolicy -> NumTxIdsToReq
maxNumTxIdsToRequest TxDecisionPolicy
pacPolicy) Key -> Key -> Key
forall a. Num a => a -> a -> a
- Key
numOfRequested)

-- | Compute the time delay until the peer should next wake to check for work.
nextWakeDelay :: PeerActionContext peeraddr txid tx
              -> Maybe DiffTime
nextWakeDelay :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> Maybe DiffTime
nextWakeDelay PeerActionContext { Time
pacNow :: forall peeraddr txid tx. PeerActionContext peeraddr txid tx -> Time
pacNow :: Time
pacNow, TxDecisionPolicy
pacPolicy :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> TxDecisionPolicy
pacPolicy :: TxDecisionPolicy
pacPolicy, DiffTime
pacClaimDelay :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> DiffTime
pacClaimDelay :: DiffTime
pacClaimDelay
                                , PeerTxLocalState tx
pacPeerState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState :: PeerTxLocalState tx
pacPeerState, PeerTxInFlight
pacPeerInFlight :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight :: PeerTxInFlight
pacPeerInFlight, SharedTxState peeraddr txid
pacSharedState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState :: SharedTxState peeraddr txid
pacSharedState } =
    (Time -> Time -> DiffTime
`diffTime` Time
pacNow) (Time -> DiffTime) -> Maybe Time -> Maybe DiffTime
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe Time -> Maybe Time -> Maybe Time
forall a. Ord a => Maybe a -> Maybe a -> Maybe a
minMaybe Maybe Time
nextClaimWake Maybe Time
nextBumpWake
  where
    -- Wake at the earliest claim-ready time among txs this peer advertises.
    nextClaimWake :: Maybe Time
nextClaimWake =
      (Maybe Time -> Key -> Maybe Time)
-> Maybe Time -> IntSet -> Maybe Time
forall a. (a -> Key -> a) -> a -> IntSet -> a
IntSet.foldl' Maybe Time -> Key -> Maybe Time
stepClaim Maybe Time
forall a. Maybe a
Nothing (PeerTxInFlight -> IntSet
pifAdvertised PeerTxInFlight
pacPeerInFlight)

    -- Wake at the earliest bump-ready time among txs this peer holds buffered.
    -- Scoped to 'peerDownloadedTxs' to match 'bumpStuckEntries': only peers
    -- with the body buffered can issue a bump, so non-buffering peers don't
    -- schedule wakes for entries they can't touch.
    nextBumpWake :: Maybe Time
nextBumpWake =
      (Maybe Time -> Key -> tx -> Maybe Time)
-> Maybe Time -> IntMap tx -> Maybe Time
forall a b. (a -> Key -> b -> a) -> a -> IntMap b -> a
IntMap.foldlWithKey' Maybe Time -> Key -> tx -> Maybe Time
forall {p}. Maybe Time -> Key -> p -> Maybe Time
stepBump Maybe Time
forall a. Maybe a
Nothing (PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
pacPeerState)

    stepClaim :: Maybe Time -> Key -> Maybe Time
stepClaim Maybe Time
acc Key
k =
      case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
pacSharedState) of
        Just TxEntry peeraddr
txEntry -> Maybe Time -> Maybe Time -> Maybe Time
forall a. Ord a => Maybe a -> Maybe a -> Maybe a
minMaybe Maybe Time
acc (TxEntry peeraddr -> Maybe Time
forall {peeraddr}. TxEntry peeraddr -> Maybe Time
futureClaimWake TxEntry peeraddr
txEntry)
        Maybe (TxEntry peeraddr)
Nothing      -> Maybe Time
acc

    stepBump :: Maybe Time -> Key -> p -> Maybe Time
stepBump Maybe Time
acc Key
k p
_tx =
      case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
pacSharedState) of
        Just TxEntry peeraddr
txEntry -> Maybe Time -> Maybe Time -> Maybe Time
forall a. Ord a => Maybe a -> Maybe a -> Maybe a
minMaybe Maybe Time
acc (Time -> TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
forall peeraddr.
Time -> TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
nextStuckBumpWake Time
pacNow TxDecisionPolicy
pacPolicy TxEntry peeraddr
txEntry)
        Maybe (TxEntry peeraddr)
Nothing      -> Maybe Time
acc

    futureClaimWake :: TxEntry peeraddr -> Maybe Time
futureClaimWake TxEntry peeraddr
txEntry =
      let claimWake :: Time
claimWake = DiffTime -> TxEntry peeraddr -> Time
forall peeraddr. DiffTime -> TxEntry peeraddr -> Time
txClaimReadyAt DiffTime
pacClaimDelay TxEntry peeraddr
txEntry in
      if Time
claimWake Time -> Time -> Bool
forall a. Ord a => a -> a -> Bool
> Time
pacNow then Time -> Maybe Time
forall a. a -> Maybe a
Just Time
claimWake
                            else Maybe Time
forall a. Maybe a
Nothing

    minMaybe :: Ord a => Maybe a -> Maybe a -> Maybe a
    minMaybe :: forall a. Ord a => Maybe a -> Maybe a -> Maybe a
minMaybe Maybe a
Nothing Maybe a
y         = Maybe a
y
    minMaybe Maybe a
x Maybe a
Nothing         = Maybe a
x
    minMaybe (Just a
x) (Just a
y) = a -> Maybe a
forall a. a -> Maybe a
Just (a -> a -> a
forall a. Ord a => a -> a -> a
min a
x a
y)

-- | Assign a tx lease to a peer and increment the attempt count.
claimTx :: peeraddr
        -> Time
        -> TxEntry peeraddr
        -> TxEntry peeraddr
claimTx :: forall peeraddr.
peeraddr -> Time -> TxEntry peeraddr -> TxEntry peeraddr
claimTx peeraddr
peeraddr Time
leaseUntil txEntry :: TxEntry peeraddr
txEntry@TxEntry { Key
txAttempt :: Key
txAttempt :: forall peeraddr. TxEntry peeraddr -> Key
txAttempt } =
  TxEntry peeraddr
txEntry {
    txLease   = TxLeased peeraddr leaseUntil,
    txAttempt = txAttempt + 1
  }

-- | Time at which a leased entry becomes eligible for an inflight cap bump.
--
-- The leaseholder's claim time is recovered from the lease deadline. If the
-- entry has been re-claimed since the original peer's attempt, this reflects
-- the latest claim instead. That is intentional: each successful claim earns
-- its own 'inflightTimeout' grace period before the next bump, so cap growth
-- is rate-limited at one bump per 'inflightTimeout' per claim.
--
-- The offset cancels the 'interTxSpace' baked into 'leaseUntil', yielding
-- 'claimTime + inflightTimeout'.
stuckBumpReadyAt :: TxDecisionPolicy -> Time -> Time
stuckBumpReadyAt :: TxDecisionPolicy -> Time -> Time
stuckBumpReadyAt TxDecisionPolicy
policy =
    DiffTime -> Time -> Time
addTime (TxDecisionPolicy -> DiffTime
inflightTimeout TxDecisionPolicy
policy DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
- TxDecisionPolicy -> DiffTime
interTxSpace TxDecisionPolicy
policy)

-- | The time at which an entry would become eligible for an inflight cap
-- bump, or 'Nothing' if the entry is not bump-eligible at all (e.g. the
-- entry is unleased, the cap is not the bottleneck, or submission is
-- already underway).
stuckBumpEligibleAt :: TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
stuckBumpEligibleAt :: forall peeraddr. TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
stuckBumpEligibleAt TxDecisionPolicy
policy
                    entry :: TxEntry peeraddr
entry@TxEntry { txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease = TxLeased peeraddr
_ Time
leaseUntil
                                  , currentMaxInflightMultiplicity :: forall peeraddr. TxEntry peeraddr -> Key
currentMaxInflightMultiplicity = Key
cap }
  | TxEntry peeraddr -> Key
forall peeraddr. TxEntry peeraddr -> Key
txAttempt TxEntry peeraddr
entry Key -> Key -> Bool
forall a. Ord a => a -> a -> Bool
>= Key
cap
  , Bool -> Bool
not (TxEntry peeraddr -> Bool
forall peeraddr. TxEntry peeraddr -> Bool
txInSubmission TxEntry peeraddr
entry)
  = Time -> Maybe Time
forall a. a -> Maybe a
Just (TxDecisionPolicy -> Time -> Time
stuckBumpReadyAt TxDecisionPolicy
policy Time
leaseUntil)
stuckBumpEligibleAt TxDecisionPolicy
_ TxEntry peeraddr
_ = Maybe Time
forall a. Maybe a
Nothing

-- | Future wake time at which an entry would become eligible for a cap bump.
--
-- Returns 'Just' for any bump-eligible entry whose 'bumpAt' is at or after
-- 'now'. Using '>=' (rather than '>') means a peer that runs 'nextWakeDelay'
-- without 'bumpStuckEntries' having already fired still schedules a wake.
nextStuckBumpWake :: Time -> TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
nextStuckBumpWake :: forall peeraddr.
Time -> TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
nextStuckBumpWake Time
now TxDecisionPolicy
policy TxEntry peeraddr
entry
  | Just Time
bumpAt <- TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
forall peeraddr. TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
stuckBumpEligibleAt TxDecisionPolicy
policy TxEntry peeraddr
entry
  , Time
bumpAt Time -> Time -> Bool
forall a. Ord a => a -> a -> Bool
>= Time
now
  = Time -> Maybe Time
forall a. a -> Maybe a
Just Time
bumpAt
nextStuckBumpWake Time
_ TxDecisionPolicy
_ TxEntry peeraddr
_ = Maybe Time
forall a. Maybe a
Nothing

-- | Sweep the txs this peer has buffered locally and bump any whose lease
-- has been held past 'inflightTimeout'. The leaseholder is in the best
-- position to detect that it is holding others up: its 'peerDownloadedTxs'
-- is small (usually empty) so the sweep is cheap, and any tx it has buffered
-- is one it has at some point claimed itself.  Note that bumps will only
-- happen for downloaded TXs.
--
-- 'sharedGeneration' is bumped when entries change so other peers wake out
-- of 'awaitSharedChange' and re-evaluate eligibility under the new cap.
bumpStuckEntries :: forall peeraddr txid tx.
                    Time
                 -> TxDecisionPolicy
                 -> IntMap.IntMap tx
                 -> SharedTxState peeraddr txid
                 -> SharedTxState peeraddr txid
bumpStuckEntries :: forall peeraddr txid tx.
Time
-> TxDecisionPolicy
-> IntMap tx
-> SharedTxState peeraddr txid
-> SharedTxState peeraddr txid
bumpStuckEntries Time
now TxDecisionPolicy
policy IntMap tx
downloadedTxs SharedTxState peeraddr txid
st =
    if Bool
anyBumped
       then SharedTxState peeraddr txid
st { sharedTxTable    = txTable',
                 sharedGeneration = sharedGeneration st + 1 }
       else SharedTxState peeraddr txid
st
  where
    (Bool
anyBumped, IntMap (TxEntry peeraddr)
txTable') =
      ((Bool, IntMap (TxEntry peeraddr))
 -> Key -> tx -> (Bool, IntMap (TxEntry peeraddr)))
-> (Bool, IntMap (TxEntry peeraddr))
-> IntMap tx
-> (Bool, IntMap (TxEntry peeraddr))
forall a b. (a -> Key -> b -> a) -> a -> IntMap b -> a
IntMap.foldlWithKey' (Bool, IntMap (TxEntry peeraddr))
-> Key -> tx -> (Bool, IntMap (TxEntry peeraddr))
forall {p}.
(Bool, IntMap (TxEntry peeraddr))
-> Key -> p -> (Bool, IntMap (TxEntry peeraddr))
bumpOne (Bool
False, SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
st) IntMap tx
downloadedTxs
    bumpOne :: (Bool, IntMap (TxEntry peeraddr))
-> Key -> p -> (Bool, IntMap (TxEntry peeraddr))
bumpOne (!Bool
changed, !IntMap (TxEntry peeraddr)
tbl) Key
k p
_tx
      | Just TxEntry peeraddr
entry  <- Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k IntMap (TxEntry peeraddr)
tbl
      , Just TxEntry peeraddr
entry' <- TxEntry peeraddr -> Maybe (TxEntry peeraddr)
bumpCurrentMaxIfStuck TxEntry peeraddr
entry
      = (Bool
True, Key
-> TxEntry peeraddr
-> IntMap (TxEntry peeraddr)
-> IntMap (TxEntry peeraddr)
forall a. Key -> a -> IntMap a -> IntMap a
IntMap.insert Key
k TxEntry peeraddr
entry' IntMap (TxEntry peeraddr)
tbl)
      | Bool
otherwise = (Bool
changed, IntMap (TxEntry peeraddr)
tbl)

    -- Bump 'currentMaxInflightMultiplicity' by one when the leaseholder
    -- has held the lease past 'inflightTimeout' without anyone reaching
    -- submission, and the cap is the bottleneck preventing another peer
    -- from joining.  Returns 'Nothing' when no bump was warranted.
    bumpCurrentMaxIfStuck :: TxEntry peeraddr -> Maybe (TxEntry peeraddr)
    bumpCurrentMaxIfStuck :: TxEntry peeraddr -> Maybe (TxEntry peeraddr)
bumpCurrentMaxIfStuck entry :: TxEntry peeraddr
entry@TxEntry { currentMaxInflightMultiplicity :: forall peeraddr. TxEntry peeraddr -> Key
currentMaxInflightMultiplicity = Key
cap }
      | Just Time
bumpAt <- TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
forall peeraddr. TxDecisionPolicy -> TxEntry peeraddr -> Maybe Time
stuckBumpEligibleAt TxDecisionPolicy
policy TxEntry peeraddr
entry
      , Time
now Time -> Time -> Bool
forall a. Ord a => a -> a -> Bool
>= Time
bumpAt
      = TxEntry peeraddr -> Maybe (TxEntry peeraddr)
forall a. a -> Maybe a
Just TxEntry peeraddr
entry { currentMaxInflightMultiplicity = cap + 1 }
    bumpCurrentMaxIfStuck TxEntry peeraddr
_ = Maybe (TxEntry peeraddr)
forall a. Maybe a
Nothing

-- | Determine if a tx is eligible for this peer to request.
--
-- A tx is selectable if it can be claimed or is already owned by this peer
-- and this peer's score-derived claim delay has elapsed.
--
-- Callers iterate candidates from the peer's own 'peerAvailableTxIds', so
-- the "this peer advertises @txKey@" precondition is established by the
-- caller.
txSelectable :: forall peeraddr txid tx.
                Eq peeraddr
             => PeerActionContext peeraddr txid tx
             -> TxKey
             -> TxEntry peeraddr
             -> Bool
txSelectable :: forall peeraddr txid tx.
Eq peeraddr =>
PeerActionContext peeraddr txid tx
-> TxKey -> TxEntry peeraddr -> Bool
txSelectable PeerActionContext { Time
pacNow :: forall peeraddr txid tx. PeerActionContext peeraddr txid tx -> Time
pacNow :: Time
pacNow, peeraddr
pacPeerAddr :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> peeraddr
pacPeerAddr :: peeraddr
pacPeerAddr, DiffTime
pacClaimDelay :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> DiffTime
pacClaimDelay :: DiffTime
pacClaimDelay
                               , PeerTxInFlight
pacPeerInFlight :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight :: PeerTxInFlight
pacPeerInFlight }
             (TxKey Key
k)
             TxEntry peeraddr
txEntry
  | TxEntry peeraddr -> Bool
forall peeraddr. TxEntry peeraddr -> Bool
txInSubmission TxEntry peeraddr
txEntry = Bool
False
  | Bool
txPeerHasAttempt = Bool
False
  | TxEntry peeraddr -> Key
forall peeraddr. TxEntry peeraddr -> Key
txAttempt TxEntry peeraddr
txEntry Key -> Key -> Bool
forall a. Ord a => a -> a -> Bool
>= TxEntry peeraddr -> Key
forall peeraddr. TxEntry peeraddr -> Key
currentMaxInflightMultiplicity TxEntry peeraddr
txEntry = Bool
False
  | TxEntry peeraddr -> Bool
txOwnedByPeer TxEntry peeraddr
txEntry = Bool
True
  | Bool
otherwise = DiffTime -> TxEntry peeraddr -> Time
forall peeraddr. DiffTime -> TxEntry peeraddr -> Time
txClaimReadyAt DiffTime
pacClaimDelay TxEntry peeraddr
txEntry Time -> Time -> Bool
forall a. Ord a => a -> a -> Bool
<= Time
pacNow
  where
    txOwnedByPeer :: TxEntry peeraddr -> Bool
    txOwnedByPeer :: TxEntry peeraddr -> Bool
txOwnedByPeer TxEntry { txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease = TxLeased peeraddr
owner Time
_ } = peeraddr
owner peeraddr -> peeraddr -> Bool
forall a. Eq a => a -> a -> Bool
== peeraddr
pacPeerAddr
    txOwnedByPeer TxEntry { txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease = TxClaimable Time
_ }    = Bool
False

    txPeerHasAttempt :: Bool
    txPeerHasAttempt :: Bool
txPeerHasAttempt =
         Key -> IntSet -> Bool
IntSet.member Key
k (PeerTxInFlight -> IntSet
pifAttempting PeerTxInFlight
pacPeerInFlight)
      Bool -> Bool -> Bool
|| Key -> IntSet -> Bool
IntSet.member Key
k (PeerTxInFlight -> IntSet
pifSubmitting PeerTxInFlight
pacPeerInFlight)

-- | Does the peer have the TX entry buffered locally?
--
-- The peer's own 'peerDownloadedTxs' is the source of truth for "buffered
-- body present", so this is a peer-local lookup.
txBufferedByPeer :: PeerTxLocalState tx -> Int -> Bool
txBufferedByPeer :: forall tx. PeerTxLocalState tx -> Key -> Bool
txBufferedByPeer PeerTxLocalState tx
peerState Key
k = Key -> IntMap tx -> Bool
forall a. Key -> IntMap a -> Bool
IntMap.member Key
k (PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
peerState)

-- | Compute the current usefulness score for a peer after time-based decay.
--
-- Scores drain at 'scoreRate' (txs/second) from the last update timestamp.
-- Returns zero for peers whose accumulated rejections have fully decayed.
currentPeerScore :: TxDecisionPolicy
                 -> Time
                 -> PeerScore
                 -> Double
currentPeerScore :: TxDecisionPolicy -> Time -> PeerScore -> Double
currentPeerScore TxDecisionPolicy { Double
scoreRate :: TxDecisionPolicy -> Double
scoreRate :: Double
scoreRate } Time
currentTime
                 PeerScore { Double
peerScoreValue :: Double
peerScoreValue :: PeerScore -> Double
peerScoreValue, Time
peerScoreTs :: Time
peerScoreTs :: PeerScore -> Time
peerScoreTs }
    | Double
peerScoreValue Double -> Double -> Bool
forall a. Eq a => a -> a -> Bool
== Double
0 = Double
0
    | Time
currentTime Time -> Time -> Bool
forall a. Ord a => a -> a -> Bool
<= Time
peerScoreTs = Double
peerScoreValue
    | Bool
otherwise = Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (Double -> Double) -> Double -> Double
forall a b. (a -> b) -> a -> b
$ Double
peerScoreValue Double -> Double -> Double
forall a. Num a => a -> a -> a
- DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (Time -> Time -> DiffTime
diffTime Time
currentTime Time
peerScoreTs) Double -> Double -> Double
forall a. Num a => a -> a -> a
* Double
scoreRate

-- | Compute the peer's score-derived claim delay.
--
-- A peer with a fully drained score has no delay.  Any non-zero score
-- maps to a delay in @[10 ms, maxPeerClaimDelay]@: the linear
-- interpolation saturates when the score reaches 'scoreMax', and the
-- 10 ms floor ensures even a small accumulated score meaningfully
-- nudges race outcomes against the peer rather than being swallowed
-- by jitter.
peerClaimDelay :: TxDecisionPolicy
               -> Time
               -> PeerScore
               -> DiffTime
peerClaimDelay :: TxDecisionPolicy -> Time -> PeerScore -> DiffTime
peerClaimDelay policy :: TxDecisionPolicy
policy@TxDecisionPolicy { DiffTime
maxPeerClaimDelay :: DiffTime
maxPeerClaimDelay :: TxDecisionPolicy -> DiffTime
maxPeerClaimDelay, Double
scoreMax :: TxDecisionPolicy -> Double
scoreMax :: Double
scoreMax } Time
currentTime PeerScore
peerScore
    | Double
s Double -> Double -> Bool
forall a. Eq a => a -> a -> Bool
== Double
0   = DiffTime
0
    | Bool
otherwise = DiffTime -> DiffTime -> DiffTime
forall a. Ord a => a -> a -> a
max DiffTime
0.010 (DiffTime -> DiffTime)
-> (Double -> DiffTime) -> Double -> DiffTime
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Double -> DiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac (Double -> DiffTime) -> Double -> DiffTime
forall a b. (a -> b) -> a -> b
$ Double
s Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
scoreMax Double -> Double -> Double
forall a. Num a => a -> a -> a
* DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
maxPeerClaimDelay
  where
    s :: Double
s = TxDecisionPolicy -> Time -> PeerScore -> Double
currentPeerScore TxDecisionPolicy
policy Time
currentTime PeerScore
peerScore

-- | Decay the peer's score to @now@, updating the timestamp.
--
-- Fast path: a score that is already 0 stays 0, and the stale
-- 'peerScoreTs' is harmless because 'currentPeerScore' short-circuits
-- on @peerScoreValue == 0@ without reading the timestamp, while any
-- later 'applyPeerEvents' transition to a positive score overwrites
-- 'peerScoreTs' with the current 'now'.  Return the state unchanged.
drainPeerScore :: TxDecisionPolicy
               -> Time
               -> PeerTxLocalState tx
               -> PeerTxLocalState tx
drainPeerScore :: forall tx.
TxDecisionPolicy
-> Time -> PeerTxLocalState tx -> PeerTxLocalState tx
drainPeerScore TxDecisionPolicy
policy Time
now peerState :: PeerTxLocalState tx
peerState@PeerTxLocalState { PeerScore
peerScore :: forall tx. PeerTxLocalState tx -> PeerScore
peerScore :: PeerScore
peerScore }
  | PeerScore -> Double
peerScoreValue PeerScore
peerScore Double -> Double -> Bool
forall a. Eq a => a -> a -> Bool
== Double
0 = PeerTxLocalState tx
peerState
  | Bool
otherwise =
      let drained :: Double
drained = TxDecisionPolicy -> Time -> PeerScore -> Double
currentPeerScore TxDecisionPolicy
policy Time
now PeerScore
peerScore in
      PeerTxLocalState tx
peerState { peerScore = PeerScore { peerScoreValue = drained, peerScoreTs = now } }
{-# INLINE drainPeerScore #-}

-- | Apply a batch of accept/reject events to the peer's local score.
-- Drains the score by elapsed time at 'scoreRate', then adds the
-- rejection count and subtracts 'scoreAcceptDecrement' per accept;
-- clamps to @[0, scoreMax]@. Returns the new score value (for
-- tracing) and the updated local state.
applyPeerEvents :: TxDecisionPolicy
                -> Time
                -> Int  -- ^ accepted count (decrements score)
                -> Int  -- ^ rejected/penalty count (increments score)
                -> PeerTxLocalState tx
                -> (Double, PeerTxLocalState tx)
applyPeerEvents :: forall tx.
TxDecisionPolicy
-> Time
-> Key
-> Key
-> PeerTxLocalState tx
-> (Double, PeerTxLocalState tx)
applyPeerEvents policy :: TxDecisionPolicy
policy@TxDecisionPolicy { Double
scoreMax :: TxDecisionPolicy -> Double
scoreMax :: Double
scoreMax, Double
scoreAcceptDecrement :: TxDecisionPolicy -> Double
scoreAcceptDecrement :: Double
scoreAcceptDecrement }
                Time
now Key
acceptedCount Key
rejectedCount
                peerState :: PeerTxLocalState tx
peerState@PeerTxLocalState { PeerScore
peerScore :: forall tx. PeerTxLocalState tx -> PeerScore
peerScore :: PeerScore
peerScore }
  -- Fast path: score is already 0 and there's no penalty to add.
  -- Accepts can only decrement (clamped at 0), so they can't move
  -- the score. Return the state unchanged.
  | Key
rejectedCount Key -> Key -> Bool
forall a. Eq a => a -> a -> Bool
== Key
0
  , PeerScore -> Double
peerScoreValue PeerScore
peerScore Double -> Double -> Bool
forall a. Eq a => a -> a -> Bool
== Double
0
  = (Double
0, PeerTxLocalState tx
peerState)

  | Bool
otherwise
  = (PeerScore -> Double
peerScoreValue PeerScore
peerScore', PeerTxLocalState tx
peerState { peerScore = peerScore' })
  where
    rejGain :: Double
rejGain = Key -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Key
rejectedCount :: Double
    accDec :: Double
accDec  = Key -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Key
acceptedCount Double -> Double -> Double
forall a. Num a => a -> a -> a
* Double
scoreAcceptDecrement
    peerScore' :: PeerScore
peerScore' =
      let !drained :: Double
drained = TxDecisionPolicy -> Time -> PeerScore -> Double
currentPeerScore TxDecisionPolicy
policy Time
now PeerScore
peerScore
          !final :: Double
final   = Double -> Double -> Double
forall a. Ord a => a -> a -> a
min Double
scoreMax (Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (Double
drained Double -> Double -> Double
forall a. Num a => a -> a -> a
+ Double
rejGain Double -> Double -> Double
forall a. Num a => a -> a -> a
- Double
accDec)) in
      PeerScore
peerScore { peerScoreValue = final, peerScoreTs = now }
{-# INLINE applyPeerEvents #-}

txClaimReadyAt :: DiffTime -> TxEntry peeraddr -> Time
txClaimReadyAt :: forall peeraddr. DiffTime -> TxEntry peeraddr -> Time
txClaimReadyAt DiffTime
claimDelay TxEntry { TxLease peeraddr
txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease :: TxLease peeraddr
txLease } =
    DiffTime -> Time -> Time
addTime DiffTime
claimDelay Time
claimableAt
  where
    claimableAt :: Time
claimableAt =
      case TxLease peeraddr
txLease of
        TxLeased peeraddr
_ Time
leaseUntil -> Time
leaseUntil
        TxClaimable Time
readyAt   -> Time
readyAt

-- | Determine if an unacknowledged txid is ready to be acknowledged.
--
-- An ack is only a count of txids to drop from the front of this peer's
-- 'peerUnacknowledgedTxIds' queue; it never refers to shared state.  So
-- once a txid is resolved it stays ackable, even after the sweep has
-- pruned both its 'sharedTxTable' entry and its retained marker.
txIdAckable :: PeerActionContext peeraddr txid tx
            -> TxKey
            -> Bool
txIdAckable :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> TxKey -> Bool
txIdAckable PeerActionContext { PeerTxLocalState tx
pacPeerState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxLocalState tx
pacPeerState :: PeerTxLocalState tx
pacPeerState, PeerTxInFlight
pacPeerInFlight :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> PeerTxInFlight
pacPeerInFlight :: PeerTxInFlight
pacPeerInFlight, SharedTxState peeraddr txid
pacSharedState :: forall peeraddr txid tx.
PeerActionContext peeraddr txid tx -> SharedTxState peeraddr txid
pacSharedState :: SharedTxState peeraddr txid
pacSharedState }
            (TxKey Key
k)
  | Key -> RetainedTxs -> Bool
retainedMember Key
k (SharedTxState peeraddr txid -> RetainedTxs
forall peeraddr txid. SharedTxState peeraddr txid -> RetainedTxs
sharedRetainedTxs SharedTxState peeraddr txid
pacSharedState) = Bool
True
    -- Resolved and still within the retention window; ack is safe.
  | Key -> IntMap tx -> Bool
forall a. Key -> IntMap a -> Bool
IntMap.member Key
k (PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
pacPeerState) = Bool
False
    -- We hold the body in our local buffer; we must submit it before
    -- acking the txid, otherwise the body would orphan in
    -- 'peerDownloadedTxs' if 'pickSubmitAction' is blocked from
    -- reaching this entry by an earlier in-flight tx.
  | Bool -> Bool
not (Key -> IntSet -> Bool
IntSet.member Key
k (PeerTxInFlight -> IntSet
pifAdvertised PeerTxInFlight
pacPeerInFlight)) = Bool
True
    -- The peer no longer tracks the txid as advertised. This covers
    -- mempool/retained txids that 'handleReceivedTxIds' kept out of the
    -- advertised set, as well as txids the peer has already attempted
    -- and submitted (or that another peer resolved).
  | Bool
otherwise =
    -- Peer still advertises the key and doesn't have the body.  Ack iff
    -- the active entry has been pruned from 'sharedTxTable' (safe late
    -- ack); otherwise the txid is still in flight and must stay
    -- unacknowledged.
    Key -> IntMap (TxEntry peeraddr) -> Bool
forall a. Key -> IntMap a -> Bool
IntMap.notMember Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
pacSharedState)

-- | Shared-state cleanup
--
-- Drops three kinds of dead entries in one pass:
--
-- * Retained entries whose retention deadline has passed.  Only the
--   'sharedRetainedTxs' membership is removed; the txid lookup tables
--   ('sharedTxIdToKey', 'sharedKeyToTxId') are preserved here because
--   peers may still hold the key in 'peerUnacknowledgedTxIds' until
--   they ack it.
-- * Orphaned 'sharedTxTable' entries: entries with a released lease,
--   no in-flight attempt, and no live peer still tracking the key.
--   These are safe to fully tear down (lookup tables included): by
--   definition no peer references them.
-- * Stale lookup-table entries: keys present only in 'sharedTxIdToKey'
--   / 'sharedKeyToTxId' with no peer still referencing them.  Bounds
--   the lookup tables so they don't grow unboundedly.
--
-- The @liveReferences@ set is the union of every active peer's
-- 'pifAdvertised' and 'pifAcksPending', snapshotted by the caller in
-- the same STM transaction as this function so it is coherent with the
-- 'sharedTxTable' read.
--
-- Bumps only 'sharedRevision' (the structural dirty bit), not
-- 'sharedGeneration' (the wake counter).  None of the three cleanup
-- categories grant other peers new options: expired-retained drops
-- and orphan removals never affect a referenced key, and stale-lookup
-- removal doesn't touch any peer-visible state.  Waking parked peers
-- here would cost a full 'nextPeerAction' pass per peer for nothing.
sweepSharedState :: HasRawTxId txid
                 => Time
                 -> IntSet
                 -> SharedTxState peeraddr txid
                 -> SharedTxState peeraddr txid
sweepSharedState :: forall txid peeraddr.
HasRawTxId txid =>
Time
-> IntSet
-> SharedTxState peeraddr txid
-> SharedTxState peeraddr txid
sweepSharedState Time
now IntSet
liveReferences SharedTxState peeraddr txid
st
    | IntSet -> Bool
IntSet.null IntSet
orphans
   Bool -> Bool -> Bool
&& IntSet -> Bool
IntSet.null IntSet
expiredRetained
   Bool -> Bool -> Bool
&& IntSet -> Bool
IntSet.null IntSet
staleLookups = SharedTxState peeraddr txid
st
    | Bool
otherwise =
        ( IntSet
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
forall {txid} {peeraddr}.
HasRawTxId txid =>
IntSet
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
dropLookupOnly IntSet
staleLookups
        (SharedTxState peeraddr txid -> SharedTxState peeraddr txid)
-> (SharedTxState peeraddr txid -> SharedTxState peeraddr txid)
-> SharedTxState peeraddr txid
-> SharedTxState peeraddr txid
forall b c a. (b -> c) -> (a -> b) -> a -> c
. IntSet
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
forall {txid} {peeraddr}.
HasRawTxId txid =>
IntSet
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
dropTxKeys IntSet
orphans
        (SharedTxState peeraddr txid -> SharedTxState peeraddr txid)
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
forall a b. (a -> b) -> a -> b
$ SharedTxState peeraddr txid
st { sharedRetainedTxs =
                 retainedDeleteKeys expiredRetained (sharedRetainedTxs st) }
        ) { sharedRevision = sharedRevision st + 1 }
  where
    expiredRetained :: IntSet
expiredRetained = Time -> RetainedTxs -> IntSet
retainedExpiredKeys Time
now (SharedTxState peeraddr txid -> RetainedTxs
forall peeraddr txid. SharedTxState peeraddr txid -> RetainedTxs
sharedRetainedTxs SharedTxState peeraddr txid
st)
    orphans :: IntSet
orphans =
      IntMap (TxEntry peeraddr) -> IntSet
forall a. IntMap a -> IntSet
IntMap.keysSet
        ((Key -> TxEntry peeraddr -> Bool)
-> IntMap (TxEntry peeraddr) -> IntMap (TxEntry peeraddr)
forall a. (Key -> a -> Bool) -> IntMap a -> IntMap a
IntMap.filterWithKey Key -> TxEntry peeraddr -> Bool
forall {peeraddr}. Key -> TxEntry peeraddr -> Bool
isOrphan (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
st))

    retainedAfter :: IntSet
retainedAfter  = RetainedTxs -> IntSet
retainedKeysSet (SharedTxState peeraddr txid -> RetainedTxs
forall peeraddr txid. SharedTxState peeraddr txid -> RetainedTxs
sharedRetainedTxs SharedTxState peeraddr txid
st)
                      IntSet -> IntSet -> IntSet
`IntSet.difference` IntSet
expiredRetained
    referencedKeys :: IntSet
referencedKeys = IntMap (TxEntry peeraddr) -> IntSet
forall a. IntMap a -> IntSet
IntMap.keysSet (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
st)
                      IntSet -> IntSet -> IntSet
`IntSet.union` IntSet
retainedAfter
                      IntSet -> IntSet -> IntSet
`IntSet.union` IntSet
liveReferences
    staleLookups :: IntSet
staleLookups   = IntMap txid -> IntSet
forall a. IntMap a -> IntSet
IntMap.keysSet (SharedTxState peeraddr txid -> IntMap txid
forall peeraddr txid. SharedTxState peeraddr txid -> IntMap txid
sharedKeyToTxId SharedTxState peeraddr txid
st)
                      IntSet -> IntSet -> IntSet
`IntSet.difference` IntSet
referencedKeys

    isOrphan :: Key -> TxEntry peeraddr -> Bool
isOrphan Key
_ TxEntry { txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease = TxLeased {} } = Bool
False
    isOrphan Key
k TxEntry { Key
txAttempt :: forall peeraddr. TxEntry peeraddr -> Key
txAttempt :: Key
txAttempt, Bool
txInSubmission :: forall peeraddr. TxEntry peeraddr -> Bool
txInSubmission :: Bool
txInSubmission }
      | Key
txAttempt Key -> Key -> Bool
forall a. Ord a => a -> a -> Bool
> Key
0                     = Bool
False
      | Bool
txInSubmission                    = Bool
False
      | Key -> IntSet -> Bool
IntSet.member Key
k IntSet
liveReferences    = Bool
False
      | Bool
otherwise                         = Bool
True

    -- Remove transaction entries from all shared state maps by key.
    dropTxKeys :: IntSet
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
dropTxKeys IntSet
keys SharedTxState peeraddr txid
s
      | IntSet -> Bool
IntSet.null IntSet
keys = SharedTxState peeraddr txid
s
      | Bool
otherwise =
          SharedTxState peeraddr txid
s {
            sharedTxTable     = IntMap.withoutKeys (sharedTxTable s) keys,
            sharedRetainedTxs = retainedDeleteKeys keys (sharedRetainedTxs s),
            sharedTxIdToKey   = IntSet.foldl' (deleteTxId (sharedKeyToTxId s))
                                              (sharedTxIdToKey s) keys,
            sharedKeyToTxId   = IntMap.withoutKeys (sharedKeyToTxId s) keys
          }

    -- Remove only the txid <-> key lookup entries.  Used for keys that
    -- have outlived their 'sharedTxTable' and retained residence but had
    -- their lookup entries kept while peers still carried the txid in
    -- 'peerUnacknowledgedTxIds'.
    dropLookupOnly :: IntSet
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
dropLookupOnly IntSet
keys SharedTxState peeraddr txid
s
      | IntSet -> Bool
IntSet.null IntSet
keys = SharedTxState peeraddr txid
s
      | Bool
otherwise =
          SharedTxState peeraddr txid
s {
            sharedTxIdToKey = IntSet.foldl' (deleteTxId (sharedKeyToTxId s))
                                            (sharedTxIdToKey s) keys,
            sharedKeyToTxId = IntMap.withoutKeys (sharedKeyToTxId s) keys
          }

    deleteTxId :: IntMap txid -> Map (RawTxId txid) a -> Key -> Map (RawTxId txid) a
deleteTxId IntMap txid
keyToTxId Map (RawTxId txid) a
txIdToKey Key
k =
      case Key -> IntMap txid -> Maybe txid
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k IntMap txid
keyToTxId of
        Just txid
txid -> RawTxId txid -> Map (RawTxId txid) a -> Map (RawTxId txid) a
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete (txid -> RawTxId txid
forall txid. HasRawTxId txid => txid -> RawTxId txid
getRawTxId txid
txid) Map (RawTxId txid) a
txIdToKey
        Maybe txid
Nothing   -> Map (RawTxId txid) a
txIdToKey
{-# INLINABLE sweepSharedState #-}

-- | Handle a batch of tx bodies received from one peer.
--
-- Received bodies are buffered locally in the peer state. Bodies that are
-- already retained or already in the mempool are counted as late and dropped.
-- Any requested tx omitted from the reply releases this peer's ownership.
-- Late TXs contributes to the returned penalty count.
handleReceivedTxs :: HasCallStack
                  => (Eq peeraddr, Show peeraddr, HasRawTxId txid)
                  => (txid -> Bool)
                  -> Time
                  -> TxDecisionPolicy
                  -> peeraddr
                  -> [(txid, tx)]
                  -> PeerTxLocalState tx
                  -> PeerTxInFlight
                  -> SharedTxState peeraddr txid
                  -> (Int, Int, PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
handleReceivedTxs :: forall peeraddr txid tx.
(HasCallStack, Eq peeraddr, Show peeraddr, HasRawTxId txid) =>
(txid -> Bool)
-> Time
-> TxDecisionPolicy
-> peeraddr
-> [(txid, tx)]
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (Key, Key, PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
handleReceivedTxs txid -> Bool
mempoolHasTx Time
now TxDecisionPolicy
policy peeraddr
peeraddr [(txid, tx)]
txs PeerTxLocalState tx
peerState PeerTxInFlight
peerInFlight SharedTxState peeraddr txid
sharedState =
    (Key
omittedCount, Key
lateCount, PeerTxLocalState tx
peerState', PeerTxInFlight
peerInFlight', SharedTxState peeraddr txid
sharedState')
  where
    txidToKey :: Map (RawTxId txid) TxKey
txidToKey = SharedTxState peeraddr txid -> Map (RawTxId txid) TxKey
forall peeraddr txid.
SharedTxState peeraddr txid -> Map (RawTxId txid) TxKey
sharedTxIdToKey SharedTxState peeraddr txid
sharedState

    requestedKeys :: IntSet
requestedKeys = RequestedTxBatch -> IntSet
requestedTxBatchSet RequestedTxBatch
requestedBatch
    retainUntil :: Time
retainUntil   = DiffTime -> Time -> Time
addTime (TxDecisionPolicy -> DiffTime
bufferedTxsMinLifetime TxDecisionPolicy
policy) Time
now

    -- Dequeue the next requested tx batch to process.
    (RequestedTxBatch
requestedBatch, StrictSeq RequestedTxBatch
remainingRequestedBatches) =
      case PeerTxLocalState tx -> StrictSeq RequestedTxBatch
forall tx. PeerTxLocalState tx -> StrictSeq RequestedTxBatch
peerRequestedTxBatches PeerTxLocalState tx
peerState of
           StrictSeq RequestedTxBatch
StrictSeq.Empty ->
             [Char] -> (RequestedTxBatch, StrictSeq RequestedTxBatch)
forall a. HasCallStack => [Char] -> a
error ([Char] -> (RequestedTxBatch, StrictSeq RequestedTxBatch))
-> [Char] -> (RequestedTxBatch, StrictSeq RequestedTxBatch)
forall a b. (a -> b) -> a -> b
$ [Char]
"TxSubmission.Inbound.V2.handleReceivedTxs: "
                  [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
"body reply received with no outstanding tx-body "
                  [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
"request from peer " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ peeraddr -> [Char]
forall a. Show a => a -> [Char]
show peeraddr
peeraddr
                  [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
" (received " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ Key -> [Char]
forall a. Show a => a -> [Char]
show ([(txid, tx)] -> Key
forall a. [a] -> Key
forall (t :: * -> *) a. Foldable t => t a -> Key
length [(txid, tx)]
txs) [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
" txs)"
           RequestedTxBatch
batch StrictSeq.:<| StrictSeq RequestedTxBatch
batches ->
             (RequestedTxBatch
batch, StrictSeq RequestedTxBatch
batches)

    -- Process each received tx, collecting late counts, pending
    -- requests, the keys still buffered by this peer, and updated
    -- shared state.  @bufferedKeys@ is the subset of @requestedKeys@
    -- that ended up buffered locally; the rest are released below.
    --
    -- @wakeChangeFromHandle@ is set only when the mempool branch fires,
    -- since its retained-insert can enable a late ack on another peer
    -- that already advertised this txid.  All other branches either
    -- leave the shared state untouched or perform a structural no-op
    -- (defensive 'IntMap.adjust' under the active/retained disjointness
    -- invariant).
    ( Key
lateCount
      , IntSet
pendingRequestedKeys
      , IntSet
bufferedKeys
      , SharedTxState peeraddr txid
sharedStateHandled
      , IntMap tx
peerDownloadedTxs'
      , Bool
wakeChangeFromHandle
      ) =
      ((Key, IntSet, IntSet, SharedTxState peeraddr txid, IntMap tx,
  Bool)
 -> (txid, tx)
 -> (Key, IntSet, IntSet, SharedTxState peeraddr txid, IntMap tx,
     Bool))
-> (Key, IntSet, IntSet, SharedTxState peeraddr txid, IntMap tx,
    Bool)
-> [(txid, tx)]
-> (Key, IntSet, IntSet, SharedTxState peeraddr txid, IntMap tx,
    Bool)
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
List.foldl'
        (Key, IntSet, IntSet, SharedTxState peeraddr txid, IntMap tx, Bool)
-> (txid, tx)
-> (Key, IntSet, IntSet, SharedTxState peeraddr txid, IntMap tx,
    Bool)
forall {a} {peeraddr} {txid} {a}.
Num a =>
(a, IntSet, IntSet, SharedTxState peeraddr txid, IntMap a, Bool)
-> (txid, a)
-> (a, IntSet, IntSet, SharedTxState peeraddr txid, IntMap a, Bool)
handleOne
        ( Key
0
        , IntSet
requestedKeys
        , IntSet
IntSet.empty
        , SharedTxState peeraddr txid
sharedState
        , PeerTxLocalState tx -> IntMap tx
forall tx. PeerTxLocalState tx -> IntMap tx
peerDownloadedTxs PeerTxLocalState tx
peerState
        , Bool
False
        )
        [(txid, tx)]
txs

    -- Process omitted (not received) txs: count a penalty for every omitted
    -- request and release this peer's lease where it still held one.
    --
    -- @wakeChangeFromOmit@ is set only when a lease is actually released
    -- (other advertisers can now claim).
    (Key
omittedCount, SharedTxState peeraddr txid
sharedStateReleased, Bool
wakeChangeFromOmit) =
      ((Key, SharedTxState peeraddr txid, Bool)
 -> Key -> (Key, SharedTxState peeraddr txid, Bool))
-> (Key, SharedTxState peeraddr txid, Bool)
-> IntSet
-> (Key, SharedTxState peeraddr txid, Bool)
forall a. (a -> Key -> a) -> a -> IntSet -> a
IntSet.foldl' (Key, SharedTxState peeraddr txid, Bool)
-> Key -> (Key, SharedTxState peeraddr txid, Bool)
forall {a} {txid}.
Num a =>
(a, SharedTxState peeraddr txid, Bool)
-> Key -> (a, SharedTxState peeraddr txid, Bool)
handleOmitted (Key
0, SharedTxState peeraddr txid
sharedStateHandled, Bool
False) IntSet
pendingRequestedKeys

    -- Keys this peer is no longer attempting (everything in the batch
    -- except still-buffered keys).  Used to update both the per-peer
    -- in-flight set and the shared 'txAttempt' counters via 'releaseLease'.
    releasedKeys :: IntSet
releasedKeys = IntSet
requestedKeys IntSet -> IntSet -> IntSet
`IntSet.difference` IntSet
bufferedKeys

    -- Bump both counters only when something happened that other peers
    -- need to react to.  In the common case (all bodies received, no
    -- mempool race) no bump fires and the structural-but-pointer-different
    -- 'sharedStateReleased' is discarded by 'writeSharedStateIfChanged'.
    sharedState' :: SharedTxState peeraddr txid
sharedState'
      | Bool
wakeChangeFromHandle Bool -> Bool -> Bool
|| Bool
wakeChangeFromOmit =
          SharedTxState peeraddr txid
sharedStateReleased {
            sharedGeneration = sharedGeneration sharedState + 1,
            sharedRevision   = sharedRevision   sharedState + 1
          }
      | Bool
otherwise =
          SharedTxState peeraddr txid
sharedState

    -- Update peer state: remove processed keys, update batch tracking,
    -- and record downloaded txs.
    peerState' :: PeerTxLocalState tx
peerState' = PeerTxLocalState tx
peerState {
        peerAvailableTxIds =
          IntMap.withoutKeys (peerAvailableTxIds peerState) requestedKeys
      , peerRequestedTxs = peerRequestedTxs peerState `IntSet.difference` requestedKeys
      , peerRequestedTxBatches = remainingRequestedBatches
      , peerRequestedTxsSize = peerRequestedTxsSize peerState - requestedTxBatchSize requestedBatch
      , peerDownloadedTxs = peerDownloadedTxs'
      }

    peerInFlight' :: PeerTxInFlight
peerInFlight' = PeerTxInFlight
peerInFlight {
        pifLeased     = pifLeased     peerInFlight `IntSet.difference` releasedKeys,
        pifAttempting = pifAttempting peerInFlight `IntSet.difference` releasedKeys,
        pifAdvertised = pifAdvertised peerInFlight `IntSet.difference` releasedKeys
      }

    keyWasLive :: Key -> Bool
keyWasLive Key
k =
         Key -> IntMap (TxEntry peeraddr) -> Bool
forall a. Key -> IntMap a -> Bool
IntMap.member Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
sharedState)
      Bool -> Bool -> Bool
|| Key -> RetainedTxs -> Bool
retainedMember Key
k (SharedTxState peeraddr txid -> RetainedTxs
forall peeraddr txid. SharedTxState peeraddr txid -> RetainedTxs
sharedRetainedTxs SharedTxState peeraddr txid
sharedState)

    -- Fold function over received txs: classify as late, already in mempool, or buffer for
    -- download.
    handleOne :: (a, IntSet, IntSet, SharedTxState peeraddr txid, IntMap a, Bool)
-> (txid, a)
-> (a, IntSet, IntSet, SharedTxState peeraddr txid, IntMap a, Bool)
handleOne
      ( !a
lateCountAcc
      , !IntSet
pendingKeysAcc
      , !IntSet
bufferedAcc
      , !SharedTxState peeraddr txid
sharedAcc
      , !IntMap a
downloadedAcc
      , !Bool
wakeAcc
      )
      (txid
txid, a
tx) =
        case RawTxId txid -> Map (RawTxId txid) TxKey -> Maybe TxKey
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup (txid -> RawTxId txid
forall txid. HasRawTxId txid => txid -> RawTxId txid
getRawTxId txid
txid) Map (RawTxId txid) TxKey
txidToKey of
             Maybe TxKey
Nothing ->
               ( a
lateCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1
               , IntSet
pendingKeysAcc
               , IntSet
bufferedAcc
               , SharedTxState peeraddr txid
sharedAcc
               , IntMap a
downloadedAcc
               , Bool
wakeAcc
               )
             Just (TxKey Key
k)
               | Key -> RetainedTxs -> Bool
retainedMember Key
k (SharedTxState peeraddr txid -> RetainedTxs
forall peeraddr txid. SharedTxState peeraddr txid -> RetainedTxs
sharedRetainedTxs SharedTxState peeraddr txid
sharedAcc) ->
                   -- Active and retained are disjoint by construction:
                   -- 'acceptSubmittedTx' deletes from 'sharedTxTable' and
                   -- inserts into 'sharedRetainedTxs' in one step, and
                   -- 'handleReceivedTxIds' leaves the active table alone
                   -- when classifying a txid as retained.  So the key is
                   -- not in 'sharedTxTable' here and there is nothing to
                   -- decrement; skip the record update.
                   ( a
lateCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1
                   , Key -> IntSet -> IntSet
IntSet.delete Key
k IntSet
pendingKeysAcc
                   , IntSet
bufferedAcc
                   , SharedTxState peeraddr txid
sharedAcc
                   , IntMap a
downloadedAcc
                   , Bool
wakeAcc
                   )
               | txid -> Bool
mempoolHasTx txid
txid ->
                   let sharedAcc' :: SharedTxState peeraddr txid
sharedAcc' =
                         SharedTxState peeraddr txid
sharedAcc {
                           sharedTxTable = IntMap.delete k (sharedTxTable sharedAcc),
                           sharedRetainedTxs =
                             retainedInsertMax k retainUntil (sharedRetainedTxs sharedAcc)
                         }
                   in ( a
lateCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1
                      , Key -> IntSet -> IntSet
IntSet.delete Key
k IntSet
pendingKeysAcc
                      , IntSet
bufferedAcc
                      , SharedTxState peeraddr txid
sharedAcc'
                      , IntMap a
downloadedAcc
                      , Bool
True
                      )
               | Bool
otherwise ->
                   case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
sharedAcc) of
                        Just TxEntry peeraddr
_txEntry
                          | Key -> IntSet -> Bool
IntSet.member Key
k (PeerTxInFlight -> IntSet
pifAttempting PeerTxInFlight
peerInFlight) ->
                              ( a
lateCountAcc
                              , Key -> IntSet -> IntSet
IntSet.delete Key
k IntSet
pendingKeysAcc
                              , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
bufferedAcc
                              , SharedTxState peeraddr txid
sharedAcc
                              , Key -> a -> IntMap a -> IntMap a
forall a. Key -> a -> IntMap a -> IntMap a
IntMap.insert Key
k a
tx IntMap a
downloadedAcc
                              , Bool
wakeAcc
                              )
                        Maybe (TxEntry peeraddr)
_ ->
                              ( a
lateCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1
                              , Key -> IntSet -> IntSet
IntSet.delete Key
k IntSet
pendingKeysAcc
                              , IntSet
bufferedAcc
                              , SharedTxState peeraddr txid
sharedAcc
                              , IntMap a
downloadedAcc
                              , Bool
wakeAcc
                              )

    -- Handle omitted (not received) txs: release this peer's lease so
    -- another advertiser can claim the tx, decrement 'txAttempt'.
    handleOmitted :: (a, SharedTxState peeraddr txid, Bool)
-> Key -> (a, SharedTxState peeraddr txid, Bool)
handleOmitted (!a
omittedCountAcc, !SharedTxState peeraddr txid
sharedAcc, !Bool
wakeAcc) Key
k
      | Key -> Bool
keyWasLive Key
k =
          case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
sharedAcc) of
            Just TxEntry peeraddr
txEntry ->
              ( a
omittedCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1
              , SharedTxState peeraddr txid
sharedAcc {
                  sharedTxTable =
                    IntMap.insert k (releaseLease txEntry)
                      (sharedTxTable sharedAcc)
                }
              , Bool
True
              )
            Maybe (TxEntry peeraddr)
Nothing ->
              (a
omittedCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1, SharedTxState peeraddr txid
sharedAcc, Bool
wakeAcc)
      | Bool
otherwise =
          (a
omittedCountAcc a -> a -> a
forall a. Num a => a -> a -> a
+ a
1, SharedTxState peeraddr txid
sharedAcc, Bool
wakeAcc)

    releaseLease :: TxEntry peeraddr -> TxEntry peeraddr
releaseLease txEntry :: TxEntry peeraddr
txEntry@TxEntry { TxLease peeraddr
txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease :: TxLease peeraddr
txLease, Key
txAttempt :: forall peeraddr. TxEntry peeraddr -> Key
txAttempt :: Key
txAttempt } =
      TxEntry peeraddr
txEntry {
        txLease = case txLease of
                       TxLeased peeraddr
owner Time
_ | peeraddr
owner peeraddr -> peeraddr -> Bool
forall a. Eq a => a -> a -> Bool
== peeraddr
peeraddr -> Time -> TxLease peeraddr
forall peeraddr. Time -> TxLease peeraddr
TxClaimable Time
now
                       TxLease peeraddr
_                                    -> TxLease peeraddr
txLease,
        txAttempt = max 0 (txAttempt - 1)
      }
{-# INLINABLE handleReceivedTxs #-}


-- | Handle the result of submitting buffered txs to the mempool.
--
-- Accepted txs leave the active table and move into the retained set so later
-- txid advertisements can be acknowledged without re-requesting the body.
-- Txs rejected by the mempool release this peer's lease and clear
-- 'txInSubmission' so another advertiser may try later.
handleSubmittedTxs :: Eq peeraddr
                   => Time
                   -> TxDecisionPolicy
                   -> peeraddr
                   -> [TxKey]
                   -> [TxKey]
                   -> PeerTxLocalState tx
                   -> PeerTxInFlight
                   -> SharedTxState peeraddr txid
                   -> (PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
handleSubmittedTxs :: forall peeraddr tx txid.
Eq peeraddr =>
Time
-> TxDecisionPolicy
-> peeraddr
-> [TxKey]
-> [TxKey]
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
handleSubmittedTxs Time
now TxDecisionPolicy
policy peeraddr
peeraddr [TxKey]
acceptedTxs [TxKey]
rejectedTxs PeerTxLocalState tx
peerState PeerTxInFlight
peerInFlight SharedTxState peeraddr txid
sharedState =
  (PeerTxLocalState tx
peerState', PeerTxInFlight
peerInFlight', SharedTxState peeraddr txid
sharedState')
  where
    acceptedKeys :: IntSet
acceptedKeys = [Key] -> IntSet
IntSet.fromList (TxKey -> Key
unTxKey (TxKey -> Key) -> [TxKey] -> [Key]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [TxKey]
acceptedTxs)
    rejectedKeys :: IntSet
rejectedKeys = [Key] -> IntSet
IntSet.fromList (TxKey -> Key
unTxKey (TxKey -> Key) -> [TxKey] -> [Key]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [TxKey]
rejectedTxs)
    submittedKeys :: IntSet
submittedKeys = IntSet
acceptedKeys IntSet -> IntSet -> IntSet
`IntSet.union` IntSet
rejectedKeys

    peerState' :: PeerTxLocalState tx
peerState' = PeerTxLocalState tx
peerState {
        peerDownloadedTxs =
          IntMap.withoutKeys (peerDownloadedTxs peerState) submittedKeys,
        peerAvailableTxIds =
          IntMap.withoutKeys (peerAvailableTxIds peerState) submittedKeys,
        -- An accepted tx from this peer is a demonstrable sign of
        -- progress: clear the empty-reply backoff flag so the txid
        -- picker may resume issuing pipelined 'ack=0, req=N' requests.
        peerLastTxIdReplyWasEmpty =
          null acceptedTxs && peerLastTxIdReplyWasEmpty peerState
      }

    -- Submission outcomes clear the peer's submission and advertised
    -- contributions for these keys; the peer is done with them.  The
    -- attempt was already taken off 'pifAttempting' (and 'txAttempt')
    -- when 'markSubmittingTxs' fired.
    peerInFlight' :: PeerTxInFlight
peerInFlight' = PeerTxInFlight
peerInFlight {
        pifLeased     = pifLeased     peerInFlight `IntSet.difference` submittedKeys,
        pifSubmitting = pifSubmitting peerInFlight `IntSet.difference` submittedKeys,
        pifAdvertised = pifAdvertised peerInFlight `IntSet.difference` submittedKeys
      }

    sharedStateAfterAccepted :: SharedTxState peeraddr txid
sharedStateAfterAccepted =
      (SharedTxState peeraddr txid
 -> TxKey -> SharedTxState peeraddr txid)
-> SharedTxState peeraddr txid
-> [TxKey]
-> SharedTxState peeraddr txid
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
List.foldl' SharedTxState peeraddr txid -> TxKey -> SharedTxState peeraddr txid
forall {peeraddr} {txid}.
SharedTxState peeraddr txid -> TxKey -> SharedTxState peeraddr txid
acceptSubmittedTx SharedTxState peeraddr txid
sharedState [TxKey]
acceptedTxs

    sharedStateAfterRejected :: SharedTxState peeraddr txid
sharedStateAfterRejected =
      (SharedTxState peeraddr txid -> Key -> SharedTxState peeraddr txid)
-> SharedTxState peeraddr txid
-> IntSet
-> SharedTxState peeraddr txid
forall a. (a -> Key -> a) -> a -> IntSet -> a
IntSet.foldl' SharedTxState peeraddr txid -> Key -> SharedTxState peeraddr txid
forall {txid}.
SharedTxState peeraddr txid -> Key -> SharedTxState peeraddr txid
updateRejected SharedTxState peeraddr txid
sharedStateAfterAccepted IntSet
rejectedKeys

    -- Both branches grant new options to other peers, so bump
    -- 'sharedGeneration': accept moves the key into 'sharedRetainedTxs'
    -- (other advertisers can ack via the retained-membership check),
    -- reject releases the lease (other advertisers can claim).
    sharedState' :: SharedTxState peeraddr txid
sharedState' =
      SharedTxState peeraddr txid
sharedStateAfterRejected {
        sharedGeneration = sharedGeneration sharedState + 1
      }

    retainedUntil :: Time
retainedUntil = DiffTime -> Time -> Time
addTime (TxDecisionPolicy -> DiffTime
bufferedTxsMinLifetime TxDecisionPolicy
policy) Time
now

    acceptSubmittedTx :: SharedTxState peeraddr txid -> TxKey -> SharedTxState peeraddr txid
acceptSubmittedTx SharedTxState peeraddr txid
sharedAcc (TxKey Key
k) =
      SharedTxState peeraddr txid
sharedAcc {
        sharedTxTable = IntMap.delete k (sharedTxTable sharedAcc),
        sharedRetainedTxs =
          retainedInsertMax k retainedUntil (sharedRetainedTxs sharedAcc)
      }

    updateRejected :: SharedTxState peeraddr txid -> Key -> SharedTxState peeraddr txid
updateRejected SharedTxState peeraddr txid
sharedAcc Key
k =
      SharedTxState peeraddr txid
sharedAcc {
        sharedTxTable = IntMap.adjust markRejected k (sharedTxTable sharedAcc)
      }

    markRejected :: TxEntry peeraddr -> TxEntry peeraddr
markRejected txEntry :: TxEntry peeraddr
txEntry@TxEntry { TxLease peeraddr
txLease :: forall peeraddr. TxEntry peeraddr -> TxLease peeraddr
txLease :: TxLease peeraddr
txLease } =
      TxEntry peeraddr
txEntry {
        txLease = case txLease of
          TxLeased peeraddr
owner Time
_ | peeraddr
owner peeraddr -> peeraddr -> Bool
forall a. Eq a => a -> a -> Bool
== peeraddr
peeraddr -> Time -> TxLease peeraddr
forall peeraddr. Time -> TxLease peeraddr
TxClaimable Time
now
          TxLease peeraddr
_                                    -> TxLease peeraddr
txLease,
        txInSubmission = False
      }
{-# INLINABLE handleSubmittedTxs #-}


-- | Mark buffered txs as entering mempool submission.
--
-- Decrements 'txAttempt' (the peer is leaving the @attempting@ state)
-- and sets 'txInSubmission'.  STM serialisation around the
-- @ChooseSubmit@ choice guarantees only one peer ever flips
-- 'txInSubmission' from @False@ to @True@ for a given key.
--
-- Bumps only 'sharedRevision' (the structural dirty bit), not
-- 'sharedGeneration' (the wake counter).  This transition only removes
-- options for other peers (they may no longer claim or submit the
-- key) and does not enable any new ack - that happens on the
-- subsequent 'handleSubmittedTxs' acceptance, which bumps the wake
-- counter itself.
markSubmittingTxs :: [TxKey]
                  -> SharedTxState peeraddr txid
                  -> SharedTxState peeraddr txid
markSubmittingTxs :: forall peeraddr txid.
[TxKey]
-> SharedTxState peeraddr txid -> SharedTxState peeraddr txid
markSubmittingTxs [] SharedTxState peeraddr txid
st = SharedTxState peeraddr txid
st
markSubmittingTxs [TxKey]
txKeys SharedTxState peeraddr txid
st =
  SharedTxState peeraddr txid
st {
    sharedTxTable  = List.foldl' markOne (sharedTxTable st) txKeys,
    sharedRevision = sharedRevision st + 1
  }
  where
    markOne :: IntMap (TxEntry peeraddr) -> TxKey -> IntMap (TxEntry peeraddr)
markOne IntMap (TxEntry peeraddr)
txTable (TxKey Key
k) = (TxEntry peeraddr -> TxEntry peeraddr)
-> Key -> IntMap (TxEntry peeraddr) -> IntMap (TxEntry peeraddr)
forall a. (a -> a) -> Key -> IntMap a -> IntMap a
IntMap.adjust TxEntry peeraddr -> TxEntry peeraddr
forall {peeraddr}. TxEntry peeraddr -> TxEntry peeraddr
markSubmitting Key
k IntMap (TxEntry peeraddr)
txTable

    markSubmitting :: TxEntry peeraddr -> TxEntry peeraddr
markSubmitting txEntry :: TxEntry peeraddr
txEntry@TxEntry { Key
txAttempt :: forall peeraddr. TxEntry peeraddr -> Key
txAttempt :: Key
txAttempt } =
      TxEntry peeraddr
txEntry {
        txAttempt      = max 0 (txAttempt - 1),
        txInSubmission = True
      }


-- | Handle a batch of txids received from one peer.
--
-- Newly seen txids are interned, appended to the peer's unacknowledged queue,
-- and entered into the shared tx table as claimable work. Any peer that later
-- advertises the txid may claim it once its score-derived delay has elapsed,
-- which avoids pinning fresh work to the first peer that happened to announce
-- it.
handleReceivedTxIds :: forall peeraddr txid tx. HasRawTxId txid
                    => (txid -> Bool)
                    -> Time
                    -> TxDecisionPolicy
                    -> NumTxIdsToReq
                    -> [(txid, SizeInBytes)]
                    -> PeerTxLocalState tx
                    -> PeerTxInFlight
                    -> SharedTxState peeraddr txid
                    -> (PeerTxLocalState tx, PeerTxInFlight, SharedTxState peeraddr txid)
handleReceivedTxIds :: forall peeraddr txid tx.
HasRawTxId txid =>
(txid -> Bool)
-> Time
-> TxDecisionPolicy
-> NumTxIdsToReq
-> [(txid, SizeInBytes)]
-> PeerTxLocalState tx
-> PeerTxInFlight
-> SharedTxState peeraddr txid
-> (PeerTxLocalState tx, PeerTxInFlight,
    SharedTxState peeraddr txid)
handleReceivedTxIds txid -> Bool
mempoolHasTx Time
now TxDecisionPolicy
policy NumTxIdsToReq
requestedTxIds [(txid, SizeInBytes)]
txidsAndSizes
                    PeerTxLocalState tx
peerState PeerTxInFlight
peerInFlight SharedTxState peeraddr txid
sharedState =
    (PeerTxLocalState tx
peerState'', PeerTxInFlight
peerInFlight'', SharedTxState peeraddr txid
sharedState'')
  where
    peerAdvertisedKeys0 :: IntSet
peerAdvertisedKeys0 = PeerTxInFlight -> IntSet
pifAdvertised PeerTxInFlight
peerInFlight
    peerAcksPending0 :: IntSet
peerAcksPending0    = PeerTxInFlight -> IntSet
pifAcksPending PeerTxInFlight
peerInFlight

    -- Fold over received txids: build unacknowledged list, update tables.
    --
    -- Two flags are accumulated:
    --   wakeChange         - mempool branch fired, whose retained-insert may
    --                        enable a late ack on another peer that already
    --                        advertised the same txid.  Bumps the wake
    --                        counter ('sharedGeneration').
    --   revisionOnlyChange - any other structural change (new entry, new
    --                        txKey interning) that other peers cannot act
    --                        on yet.  Bumps only the dirty bit
    --                        ('sharedRevision').
    ( [TxKey]
receivedTxKeysRev
      , IntMap SizeInBytes
peerAvailableTxIds'
      , SharedTxState peeraddr txid
sharedStateHandled
      , IntSet
peerAdvertisedKeys'
      , IntSet
peerAcksPending'
      , Bool
wakeChange
      , Bool
revisionOnlyChange
      ) =
      (([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid, IntSet,
  IntSet, Bool, Bool)
 -> (txid, SizeInBytes)
 -> ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid,
     IntSet, IntSet, Bool, Bool))
-> ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid,
    IntSet, IntSet, Bool, Bool)
-> [(txid, SizeInBytes)]
-> ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid,
    IntSet, IntSet, Bool, Bool)
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
List.foldl'
        ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid, IntSet,
 IntSet, Bool, Bool)
-> (txid, SizeInBytes)
-> ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid,
    IntSet, IntSet, Bool, Bool)
step
        ( []
        , PeerTxLocalState tx -> IntMap SizeInBytes
forall tx. PeerTxLocalState tx -> IntMap SizeInBytes
peerAvailableTxIds PeerTxLocalState tx
peerState
        , SharedTxState peeraddr txid
sharedState
        , IntSet
peerAdvertisedKeys0
        , IntSet
peerAcksPending0
        , Bool
False
        , Bool
False
        )
        [(txid, SizeInBytes)]
txidsAndSizes

    peerUnacknowledgedTxIds' :: StrictSeq TxKey
peerUnacknowledgedTxIds' =
      PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
peerState StrictSeq TxKey -> StrictSeq TxKey -> StrictSeq TxKey
forall a. Semigroup a => a -> a -> a
<> [TxKey] -> StrictSeq TxKey
forall a. [a] -> StrictSeq a
StrictSeq.fromList ([TxKey] -> [TxKey]
forall a. [a] -> [a]
reverse [TxKey]
receivedTxKeysRev)

    peerState'' :: PeerTxLocalState tx
peerState'' = PeerTxLocalState tx
peerState {
        peerUnacknowledgedTxIds = peerUnacknowledgedTxIds',
        peerRequestedTxIds =
          assert (requestedTxIds <= peerRequestedTxIds peerState)
                 (peerRequestedTxIds peerState - requestedTxIds),
        peerAvailableTxIds = peerAvailableTxIds',
        -- Track empty-reply state so the txid picker can back off
        -- 'ack=0, req=N' pipelined requests until the peer makes
        -- progress.
        peerLastTxIdReplyWasEmpty = null txidsAndSizes
      }

    -- An empty reply leaves both sets unchanged (the fold doesn't fire),
    -- so skip the record rebuild.
    peerInFlight'' :: PeerTxInFlight
peerInFlight'' =
        if [(txid, SizeInBytes)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(txid, SizeInBytes)]
txidsAndSizes
           then PeerTxInFlight
peerInFlight
           else PeerTxInFlight
peerInFlight {
               pifAdvertised  = peerAdvertisedKeys',
               pifAcksPending = peerAcksPending'
             }

    sharedState'' :: SharedTxState peeraddr txid
sharedState''
      | Bool
wakeChange =
          SharedTxState peeraddr txid
sharedStateHandled {
            sharedGeneration = sharedGeneration sharedState + 1,
            sharedRevision   = sharedRevision   sharedState + 1
          }
      | Bool
revisionOnlyChange =
          SharedTxState peeraddr txid
sharedStateHandled {
            sharedRevision = sharedRevision sharedState + 1
          }
      | Bool
otherwise =
          SharedTxState peeraddr txid
sharedState

    retainUntil :: Time
retainUntil = DiffTime -> Time -> Time
addTime (TxDecisionPolicy -> DiffTime
bufferedTxsMinLifetime TxDecisionPolicy
policy) Time
now

    -- Process each received txid: classify as retained, in mempool, or new entry.
    step
      :: ( [TxKey]
         , IntMap.IntMap SizeInBytes
         , SharedTxState peeraddr txid
         , IntSet.IntSet
         , IntSet.IntSet
         , Bool                     -- wakeChange
         , Bool                     -- revisionOnlyChange
         )
      -> (txid, SizeInBytes)
      -> ( [TxKey]
         , IntMap.IntMap SizeInBytes
         , SharedTxState peeraddr txid
         , IntSet.IntSet
         , IntSet.IntSet
         , Bool
         , Bool
         )
    step :: ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid, IntSet,
 IntSet, Bool, Bool)
-> (txid, SizeInBytes)
-> ([TxKey], IntMap SizeInBytes, SharedTxState peeraddr txid,
    IntSet, IntSet, Bool, Bool)
step
      ( ![TxKey]
unacknowledgedAcc
      , !IntMap SizeInBytes
availableAcc
      , !SharedTxState peeraddr txid
sharedAcc
      , !IntSet
peerAdvertisedKeysAcc
      , !IntSet
peerAcksPendingAcc
      , !Bool
wakeAcc
      , !Bool
revOnlyAcc
      )
      (txid
txid, SizeInBytes
txSize)
      | Key -> RetainedTxs -> Bool
retainedMember Key
k RetainedTxs
retainedAcc =
          -- No 'sharedTxTable' change.  Only revision-relevant if interning
          -- added a new key.
          ( TxKey
txKey TxKey -> [TxKey] -> [TxKey]
forall a. a -> [a] -> [a]
: [TxKey]
unacknowledgedAcc
          , Key -> IntMap SizeInBytes -> IntMap SizeInBytes
forall a. Key -> IntMap a -> IntMap a
IntMap.delete Key
k IntMap SizeInBytes
availableAcc
          , SharedTxState peeraddr txid
sharedAcc'
          , Key -> IntSet -> IntSet
IntSet.delete Key
k IntSet
peerAdvertisedKeysAcc
          , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
peerAcksPendingAcc
          , Bool
wakeAcc
          , Bool
revOnlyAcc'
          )
      | txid -> Bool
mempoolHasTx txid
txid =
          -- Retained-insert may enable a late ack on another peer that
          -- already advertised this txid: wake-relevant.
          ( TxKey
txKey TxKey -> [TxKey] -> [TxKey]
forall a. a -> [a] -> [a]
: [TxKey]
unacknowledgedAcc
          , Key -> IntMap SizeInBytes -> IntMap SizeInBytes
forall a. Key -> IntMap a -> IntMap a
IntMap.delete Key
k IntMap SizeInBytes
availableAcc
          , SharedTxState peeraddr txid
sharedAcc' {
              sharedTxTable = IntMap.delete k (sharedTxTable sharedAcc'),
              sharedRetainedTxs =
                retainedInsertMax k retainUntil (sharedRetainedTxs sharedAcc')
            }
          , Key -> IntSet -> IntSet
IntSet.delete Key
k IntSet
peerAdvertisedKeysAcc
          , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
peerAcksPendingAcc
          , Bool
True
          , Bool
True
          )
      | Bool
otherwise =
          case Key -> IntMap (TxEntry peeraddr) -> Maybe (TxEntry peeraddr)
forall a. Key -> IntMap a -> Maybe a
IntMap.lookup Key
k (SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
forall peeraddr txid.
SharedTxState peeraddr txid -> IntMap (TxEntry peeraddr)
sharedTxTable SharedTxState peeraddr txid
sharedAcc') of
            Maybe (TxEntry peeraddr)
Nothing ->
              -- New 'TxClaimable' entry.  Only this peer knows about it,
              -- so no wake; revision tracks the structural change.
              let txEntry :: TxEntry peeraddr
txEntry = TxEntry {
                              txLease :: TxLease peeraddr
txLease = Time -> TxLease peeraddr
forall peeraddr. Time -> TxLease peeraddr
TxClaimable Time
now,
                              txAttempt :: Key
txAttempt = Key
0,
                              txInSubmission :: Bool
txInSubmission = Bool
False,
                              currentMaxInflightMultiplicity :: Key
currentMaxInflightMultiplicity =
                                TxDecisionPolicy -> Key
txInflightMultiplicity TxDecisionPolicy
policy
                            }
              in ( TxKey
txKey TxKey -> [TxKey] -> [TxKey]
forall a. a -> [a] -> [a]
: [TxKey]
unacknowledgedAcc
                 , Key -> SizeInBytes -> IntMap SizeInBytes -> IntMap SizeInBytes
forall a. Key -> a -> IntMap a -> IntMap a
IntMap.insert Key
k SizeInBytes
txSize IntMap SizeInBytes
availableAcc
                 , SharedTxState peeraddr txid
sharedAcc' {
                     sharedTxTable = IntMap.insert k txEntry (sharedTxTable sharedAcc')
                   }
                 , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
peerAdvertisedKeysAcc
                 , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
peerAcksPendingAcc
                 , Bool
wakeAcc
                 , Bool
True
                 )
            Just TxEntry peeraddr
_ ->
              -- Existing entry; another peer already advertised.  No
              -- 'sharedTxTable' change beyond optional interning.
              ( TxKey
txKey TxKey -> [TxKey] -> [TxKey]
forall a. a -> [a] -> [a]
: [TxKey]
unacknowledgedAcc
              , Key -> SizeInBytes -> IntMap SizeInBytes -> IntMap SizeInBytes
forall a. Key -> a -> IntMap a -> IntMap a
IntMap.insert Key
k SizeInBytes
txSize IntMap SizeInBytes
availableAcc
              , SharedTxState peeraddr txid
sharedAcc'
              , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
peerAdvertisedKeysAcc
              , Key -> IntSet -> IntSet
IntSet.insert Key
k IntSet
peerAcksPendingAcc
              , Bool
wakeAcc
              , Bool
revOnlyAcc'
              )
      where
        retainedAcc :: RetainedTxs
retainedAcc = SharedTxState peeraddr txid -> RetainedTxs
forall peeraddr txid. SharedTxState peeraddr txid -> RetainedTxs
sharedRetainedTxs SharedTxState peeraddr txid
sharedAcc'
        revOnlyAcc' :: Bool
revOnlyAcc' = Bool
revOnlyAcc Bool -> Bool -> Bool
|| Bool
txKeyWasNew
        (txKey :: TxKey
txKey@(TxKey Key
k), Bool
txKeyWasNew, SharedTxState peeraddr txid
sharedAcc') = txid
-> SharedTxState peeraddr txid
-> (TxKey, Bool, SharedTxState peeraddr txid)
forall {txid} {peeraddr}.
HasRawTxId txid =>
txid
-> SharedTxState peeraddr txid
-> (TxKey, Bool, SharedTxState peeraddr txid)
lookupOrInternTxId txid
txid SharedTxState peeraddr txid
sharedAcc

    lookupOrInternTxId :: txid
-> SharedTxState peeraddr txid
-> (TxKey, Bool, SharedTxState peeraddr txid)
lookupOrInternTxId txid
txid SharedTxState peeraddr txid
st =
      let (RawTxId txid
_, TxKey
key, SharedTxState peeraddr txid
st') = txid
-> SharedTxState peeraddr txid
-> (RawTxId txid, TxKey, SharedTxState peeraddr txid)
forall txid peeraddr.
HasRawTxId txid =>
txid
-> SharedTxState peeraddr txid
-> (RawTxId txid, TxKey, SharedTxState peeraddr txid)
internTxId txid
txid SharedTxState peeraddr txid
st
      in (TxKey
key, SharedTxState peeraddr txid -> Key
forall peeraddr txid. SharedTxState peeraddr txid -> Key
sharedNextTxKey SharedTxState peeraddr txid
st' Key -> Key -> Bool
forall a. Eq a => a -> a -> Bool
/= SharedTxState peeraddr txid -> Key
forall peeraddr txid. SharedTxState peeraddr txid -> Key
sharedNextTxKey SharedTxState peeraddr txid
st, SharedTxState peeraddr txid
st')
{-# INLINABLE handleReceivedTxIds #-}