{-# LANGUAGE BangPatterns        #-}
{-# LANGUAGE DataKinds           #-}
{-# LANGUAGE GADTs               #-}
{-# LANGUAGE KindSignatures      #-}
{-# LANGUAGE LambdaCase          #-}
{-# LANGUAGE NamedFieldPuns      #-}
{-# LANGUAGE ScopedTypeVariables #-}

module Ouroboros.Network.TxSubmission.Inbound.V2
  ( -- * TxSubmission Inbound client
    txSubmissionInboundV2
    -- * Supporting types and APIs
  , module V2
  , TxDecisionPolicy (..)
  , defaultTxDecisionPolicy
  , TxSubmissionInitDelay (..)
  ) where

import Data.List.NonEmpty qualified as NonEmpty
import Data.Map.Strict qualified as Map
import Data.Maybe (mapMaybe)
import Data.Sequence.Strict qualified as StrictSeq
import Data.Set qualified as Set
import Data.Typeable (Typeable)

import Control.Concurrent.Class.MonadSTM.Strict
import Control.Monad (unless)
import Control.Monad.Class.MonadThrow
import Control.Monad.Class.MonadTime.SI
import Control.Monad.Class.MonadTimer.SI
import Control.Tracer (Tracer, traceWith)

import Network.TypedProtocol

import Ouroboros.Network.Protocol.TxSubmission2.Server
import Ouroboros.Network.Protocol.TxSubmission2.Type (NumTxIdsToAck,
           SizeInBytes)
import Ouroboros.Network.TxSubmission.Inbound.V2.Policy
import Ouroboros.Network.TxSubmission.Inbound.V2.Registry as V2
import Ouroboros.Network.TxSubmission.Inbound.V2.State qualified as State
import Ouroboros.Network.TxSubmission.Inbound.V2.Types as V2

-- The same Stateful types as V1 uses.
newtype Stateful s n txid tx m = Stateful (s -> ServerStIdle n txid tx m ())

newtype StatefulM s n txid tx m
  = StatefulM (s -> m (ServerStIdle n txid tx m ()))

newtype StatefulCollect s n txid tx m
  = StatefulCollect (s -> Collect txid tx -> m (ServerStIdle n txid tx m ()))

continueWithState :: Stateful s n txid tx m
                  -> s
                  -> ServerStIdle n txid tx m ()
continueWithState :: forall s (n :: N) txid tx (m :: * -> *).
Stateful s n txid tx m -> s -> ServerStIdle n txid tx m ()
continueWithState (Stateful s -> ServerStIdle n txid tx m ()
f) !s
st =
    s -> ServerStIdle n txid tx m ()
f s
st
{-# INLINE continueWithState #-}

continueWithStateM :: StatefulM s n txid tx m
                   -> s
                   -> m (ServerStIdle n txid tx m ())
continueWithStateM :: forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (StatefulM s -> m (ServerStIdle n txid tx m ())
f) !s
st =
    s -> m (ServerStIdle n txid tx m ())
f s
st
{-# INLINE continueWithStateM #-}

collectAndContinueWithState :: StatefulCollect s n txid tx m
                            -> s
                            -> Collect txid tx
                            -> m (ServerStIdle n txid tx m ())
collectAndContinueWithState :: forall s (n :: N) txid tx (m :: * -> *).
StatefulCollect s n txid tx m
-> s -> Collect txid tx -> m (ServerStIdle n txid tx m ())
collectAndContinueWithState (StatefulCollect s -> Collect txid tx -> m (ServerStIdle n txid tx m ())
f) !s
st =
    s -> Collect txid tx -> m (ServerStIdle n txid tx m ())
f s
st
{-# INLINE collectAndContinueWithState #-}

-- | A tx-submission inbound side (server, sic!).
--
-- Each call to 'runNextPeerAction' atomically inspects the shared state and
-- this peer's local protocol state and returns one of: submit buffered
-- bodies to the mempool, request bodies, request txids (blocking or
-- pipelined), or do nothing.  When idle, the server parks on
-- 'awaitSharedChange' until the shared-state generation moves or an
-- optional wake delay expires, then re-evaluates.  Body requests are
-- pipelined; in pipelined mode txid requests become non-blocking and the
-- server collects replies via 'handleReplies'.
--
-- V2 server state machine.
--
-- Depth 'n' tracks outstanding pipelined replies (type-level Nat).
--
-- States (non-pipelined, n = 0):
--   serverIdle           - park on awaitSharedChange or pick next action
--   serverReqTxIds 0     - send a txid request (blocking or pipelined)
--   submitBufferedTxs    - submit buffered bodies, run continuation
--   requestTxBodies      - send a pipelined body request, depth + 1
--
-- States (pipelined, n > 0):
--   continueAfterReplies      - post-reply pick; dispatch action or DoNothing
--   continueAfterBodyRequests - post-body-request pick; same dispatch
--   handleReplies n           - block on CollectPipelined for the next reply
--   handleReply  (n-1)        - apply received txids or bodies, then
--                               continueAfterReplies (n-1)
--
-- Transitions:
--
--   serverIdle
--     [PeerDoNothing]       --> awaitSharedChange   --> serverIdle
--     [PeerSubmitTxs ks]    --> submitBufferedTxs ks serverIdle
--     [PeerRequestTxs ks]   --> requestTxBodies 0 ks
--                                 --> continueAfterBodyRequests 1
--     [PeerRequestTxIds] when unack queue empty --> blocking wire request,
--                                                    wait for reply --> serverIdle
--     [PeerRequestTxIds] otherwise              --> pipelined wire request
--                                                    --> handleReplies 1
--
--   handleReplies n --> handleReply (n-1)
--   handleReply
--     [CollectTxIds] --> applyReceivedTxIds --> continueAfterReplies (n-1)
--     [CollectTxs]   --> applyReceivedTxs   --> continueAfterReplies (n-1)
--
--   continueAfterReplies 0 = serverIdle
--   continueAfterReplies n@(>0), continueAfterBodyRequests n
--     [PeerSubmitTxs ks]    --> submitBufferedTxs ks (continueAfterReplies n)
--     [PeerRequestTxs ks]   --> requestTxBodies n ks
--     [PeerRequestTxIds]    --> serverReqTxIds n
--     [PeerDoNothing]       --> handleReplies n
txSubmissionInboundV2
  :: forall txid tx idx m err.
     ( MonadDelay m
     , MonadSTM m
     , MonadThrow m
     , Ord txid
     , Show txid
     , Typeable txid
     )
  => Tracer m (TraceTxSubmissionInbound txid tx)
  -> TxSubmissionInitDelay
  -> TxDecisionPolicy
  -> TxSubmissionMempoolWriter txid tx idx m err
  -> (tx -> SizeInBytes)
  -> PeerTxAPI m txid tx
  -> TxSubmissionServerPipelined txid tx m ()
txSubmissionInboundV2 :: forall txid tx idx (m :: * -> *) err.
(MonadDelay m, MonadSTM m, MonadThrow m, Ord txid, Show txid,
 Typeable txid) =>
Tracer m (TraceTxSubmissionInbound txid tx)
-> TxSubmissionInitDelay
-> TxDecisionPolicy
-> TxSubmissionMempoolWriter txid tx idx m err
-> (tx -> SizeInBytes)
-> PeerTxAPI m txid tx
-> TxSubmissionServerPipelined txid tx m ()
txSubmissionInboundV2
    Tracer m (TraceTxSubmissionInbound txid tx)
tracer
    TxSubmissionInitDelay
initDelay
    TxDecisionPolicy
policy
    TxSubmissionMempoolWriter { tx -> txid
txId :: tx -> txid
txId :: forall txid tx idx (m :: * -> *) err.
TxSubmissionMempoolWriter txid tx idx m err -> tx -> txid
txId, [tx] -> m ([txid], [(txid, err)])
mempoolAddTxs :: [tx] -> m ([txid], [(txid, err)])
mempoolAddTxs :: forall txid tx idx (m :: * -> *) err.
TxSubmissionMempoolWriter txid tx idx m err
-> [tx] -> m ([txid], [(txid, err)])
mempoolAddTxs }
    tx -> SizeInBytes
txSize
    PeerTxAPI {
      Word64 -> Maybe DiffTime -> m ()
awaitSharedChange :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx -> Word64 -> Maybe DiffTime -> m ()
awaitSharedChange :: Word64 -> Maybe DiffTime -> m ()
awaitSharedChange,
      Time -> PeerTxLocalState tx -> m (PeerAction, PeerTxLocalState tx)
runNextPeerAction :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> Time
-> PeerTxLocalState tx
-> m (PeerAction, PeerTxLocalState tx)
runNextPeerAction :: Time -> PeerTxLocalState tx -> m (PeerAction, PeerTxLocalState tx)
runNextPeerAction,
      Time -> PeerTxLocalState tx -> m (PeerAction, PeerTxLocalState tx)
runNextPeerActionPipelined :: Time -> PeerTxLocalState tx -> m (PeerAction, PeerTxLocalState tx)
runNextPeerActionPipelined :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> Time
-> PeerTxLocalState tx
-> m (PeerAction, PeerTxLocalState tx)
runNextPeerActionPipelined,
      Time
-> NumTxIdsToReq
-> [(txid, SizeInBytes)]
-> PeerTxLocalState tx
-> m (PeerTxLocalState tx)
applyReceivedTxIds :: Time
-> NumTxIdsToReq
-> [(txid, SizeInBytes)]
-> PeerTxLocalState tx
-> m (PeerTxLocalState tx)
applyReceivedTxIds :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> Time
-> NumTxIdsToReq
-> [(txid, SizeInBytes)]
-> PeerTxLocalState tx
-> m (PeerTxLocalState tx)
applyReceivedTxIds,
      Time
-> [(txid, tx)]
-> PeerTxLocalState tx
-> m (Int, PeerTxLocalState tx)
applyReceivedTxs :: Time
-> [(txid, tx)]
-> PeerTxLocalState tx
-> m (Int, PeerTxLocalState tx)
applyReceivedTxs :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> Time
-> [(txid, tx)]
-> PeerTxLocalState tx
-> m (Int, PeerTxLocalState tx)
applyReceivedTxs,
      Time
-> [TxKey]
-> [TxKey]
-> PeerTxLocalState tx
-> m (PeerTxLocalState tx)
applySubmittedTxs :: Time
-> [TxKey]
-> [TxKey]
-> PeerTxLocalState tx
-> m (PeerTxLocalState tx)
applySubmittedTxs :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> Time
-> [TxKey]
-> [TxKey]
-> PeerTxLocalState tx
-> m (PeerTxLocalState tx)
applySubmittedTxs,
      PeerTxLocalState tx -> [TxKey] -> m (Map txid SizeInBytes)
resolveTxRequest :: PeerTxLocalState tx -> [TxKey] -> m (Map txid SizeInBytes)
resolveTxRequest :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> PeerTxLocalState tx -> [TxKey] -> m (Map txid SizeInBytes)
resolveTxRequest,
      PeerTxLocalState tx -> [TxKey] -> m [(TxKey, txid, tx)]
resolveBufferedTxs :: PeerTxLocalState tx -> [TxKey] -> m [(TxKey, txid, tx)]
resolveBufferedTxs :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx
-> PeerTxLocalState tx -> [TxKey] -> m [(TxKey, txid, tx)]
resolveBufferedTxs,
      TxSubmissionCounters -> m ()
addCounters :: TxSubmissionCounters -> m ()
addCounters :: forall (m :: * -> *) txid tx.
PeerTxAPI m txid tx -> TxSubmissionCounters -> m ()
addCounters
    } =
    m (ServerStIdle 'Z txid tx m ())
-> TxSubmissionServerPipelined txid tx m ()
forall (m :: * -> *) txid tx a.
m (ServerStIdle 'Z txid tx m a)
-> TxSubmissionServerPipelined txid tx m a
TxSubmissionServerPipelined (m (ServerStIdle 'Z txid tx m ())
 -> TxSubmissionServerPipelined txid tx m ())
-> m (ServerStIdle 'Z txid tx m ())
-> TxSubmissionServerPipelined txid tx m ()
forall a b. (a -> b) -> a -> b
$ do
      case TxSubmissionInitDelay
initDelay of
        TxSubmissionInitDelay DiffTime
delay -> DiffTime -> m ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
delay
        TxSubmissionInitDelay
NoTxSubmissionInitDelay     -> () -> m ()
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
      StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle PeerTxLocalState tx
forall tx. PeerTxLocalState tx
emptyPeerTxLocalState
  where

    -- Mirror V1's TraceTxInboundCan/CannotRequestMoreTxs: emit once per
    -- scheduler iteration based on whether the peer currently advertises any
    -- txids whose bodies we could request.
    traceCanRequest :: forall (n :: N). Nat n -> PeerTxLocalState tx -> m ()
    traceCanRequest :: forall (n :: N). Nat n -> PeerTxLocalState tx -> m ()
traceCanRequest Nat n
n PeerTxLocalState tx
st
      | IntMap SizeInBytes -> Bool
forall a. IntMap a -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null (PeerTxLocalState tx -> IntMap SizeInBytes
forall tx. PeerTxLocalState tx -> IntMap SizeInBytes
peerAvailableTxIds PeerTxLocalState tx
st) =
          Tracer m (TraceTxSubmissionInbound txid tx)
-> TraceTxSubmissionInbound txid tx -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceTxSubmissionInbound txid tx)
tracer (Int -> TraceTxSubmissionInbound txid tx
forall txid tx. Int -> TraceTxSubmissionInbound txid tx
TraceTxInboundCannotRequestMoreTxs (Nat n -> Int
forall (n :: N). Nat n -> Int
natToInt Nat n
n))
      | Bool
otherwise =
          Tracer m (TraceTxSubmissionInbound txid tx)
-> TraceTxSubmissionInbound txid tx -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceTxSubmissionInbound txid tx)
tracer (Int -> TraceTxSubmissionInbound txid tx
forall txid tx. Int -> TraceTxSubmissionInbound txid tx
TraceTxInboundCanRequestMoreTxs (Nat n -> Int
forall (n :: N). Nat n -> Int
natToInt Nat n
n))

    -- Entry point and reset state for the non-pipelined server loop.
    --
    -- This function is called when:
    --   1. The server first starts
    --   2. All pipelined requests have completed and the counter returns to zero
    --   3. An idle peer wakes up after a @PeerDoNothing@ wait
    serverIdle :: StatefulM (PeerTxLocalState tx) Z txid tx m
    serverIdle :: StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle = (PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ()))
-> StatefulM (PeerTxLocalState tx) 'Z txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) 'Z txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ()))
-> StatefulM (PeerTxLocalState tx) 'Z txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> do
      now <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
      -- When the pipeline fully drains, emit the body-download episode
      -- duration (covers all overlapping body and txid pipelined requests).
      peerState' <- case peerDownloadStartTime peerState of
                         Maybe Time
Nothing        -> PeerTxLocalState tx -> m (PeerTxLocalState tx)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure PeerTxLocalState tx
peerState
                         Just Time
startTime -> do
                           TxSubmissionCounters -> m ()
addCounters TxSubmissionCounters
forall a. Monoid a => a
mempty { txPipelineWaitMs =
                                                  diffTimeToMilliseconds (now `diffTime` startTime) }
                           PeerTxLocalState tx -> m (PeerTxLocalState tx)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (PeerTxLocalState tx -> m (PeerTxLocalState tx))
-> PeerTxLocalState tx -> m (PeerTxLocalState tx)
forall a b. (a -> b) -> a -> b
$ PeerTxLocalState tx
peerState { peerDownloadStartTime = Nothing }
      traceCanRequest Zero peerState'
      (peerAction, peerState'') <- runNextPeerAction now (State.drainPeerScore policy now peerState')
      case peerAction of
           PeerDoNothing Word64
generation Maybe DiffTime
mDelay -> do
             -- An Active->Idle transition means this peer has just become
             -- eligible for actions it could not take before (e.g. claiming
             -- expired leases as an idle claimant). Re-run the scheduler
             -- immediately rather than parking on a wake condition that may
             -- not fire.
             let cameToIdle :: Bool
cameToIdle = PeerTxLocalState tx -> PeerPhase
forall tx. PeerTxLocalState tx -> PeerPhase
peerPhase PeerTxLocalState tx
peerState' PeerPhase -> PeerPhase -> Bool
forall a. Eq a => a -> a -> Bool
/= PeerPhase
PeerIdle
                           Bool -> Bool -> Bool
&& PeerTxLocalState tx -> PeerPhase
forall tx. PeerTxLocalState tx -> PeerPhase
peerPhase PeerTxLocalState tx
peerState'' PeerPhase -> PeerPhase -> Bool
forall a. Eq a => a -> a -> Bool
== PeerPhase
PeerIdle
             if Bool
cameToIdle
                then StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle PeerTxLocalState tx
peerState''
                else do
                  Word64 -> Maybe DiffTime -> m ()
awaitSharedChange Word64
generation Maybe DiffTime
mDelay
                  StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle PeerTxLocalState tx
peerState''
           PeerSubmitTxs [TxKey]
txKeys ->
             StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM ([TxKey]
-> StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> StatefulM (PeerTxLocalState tx) 'Z txid tx m
forall (n :: N).
[TxKey]
-> StatefulM (PeerTxLocalState tx) n txid tx m
-> StatefulM (PeerTxLocalState tx) n txid tx m
submitBufferedTxs [TxKey]
txKeys StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle) PeerTxLocalState tx
peerState''
           PeerRequestTxs [TxKey]
txKeys ->
             StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (Nat 'Z -> [TxKey] -> StatefulM (PeerTxLocalState tx) 'Z txid tx m
forall (n :: N).
Nat n -> [TxKey] -> StatefulM (PeerTxLocalState tx) n txid tx m
requestTxBodies Nat 'Z
forall (n :: N). ('Z ~ n) => Nat n
Zero [TxKey]
txKeys) PeerTxLocalState tx
peerState''
           PeerRequestTxIds NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq ->
             StatefulM (PeerTxLocalState tx) 'Z txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle 'Z txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (Nat 'Z
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) 'Z txid tx m
forall (n :: N).
Nat n
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) n txid tx m
serverReqTxIds Nat 'Z
forall (n :: N). ('Z ~ n) => Nat n
Zero NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq) PeerTxLocalState tx
peerState''

    -- | Submit buffered transaction bodies to the mempool.
    submitBufferedTxs :: forall (n :: N).
                         [TxKey]
                      -> StatefulM (PeerTxLocalState tx) n txid tx m
                      -- ^ a continuation
                      -> StatefulM (PeerTxLocalState tx) n txid tx m
    submitBufferedTxs :: forall (n :: N).
[TxKey]
-> StatefulM (PeerTxLocalState tx) n txid tx m
-> StatefulM (PeerTxLocalState tx) n txid tx m
submitBufferedTxs [TxKey]
txKeys StatefulM (PeerTxLocalState tx) n txid tx m
k = (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> do
      bufferedTxs <- PeerTxLocalState tx -> [TxKey] -> m [(TxKey, txid, tx)]
resolveBufferedTxs PeerTxLocalState tx
peerState [TxKey]
txKeys

      start <- getMonotonicTime
      let submitted = [ (TxKey
txKey, txid
txid') | (TxKey
txKey, txid
txid', tx
_) <- [(TxKey, txid, tx)]
bufferedTxs ]
          toSubmit  = [ tx
tx | (TxKey
_, txid
_, tx
tx) <- [(TxKey, txid, tx)]
bufferedTxs ]

      (acceptedTxIds, rejectedTxs) <- if null toSubmit
                                         then pure ([], [])
                                         else mempoolAddTxs toSubmit
      end <- getMonotonicTime

      -- 'mempoolAddTxs' partitions the batch into accepted and
      -- rejected (see 'TxSubmissionMempoolWriter'); map each verdict's
      -- txids back to our 'TxKey's via the submitted batch.
      let submittedKeyOf   = [(txid, TxKey)] -> Map txid TxKey
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList [ (txid
txid', TxKey
txKey) | (TxKey
txKey, txid
txid') <- [(TxKey, txid)]
submitted ]
          resolvedTxKeys   = (txid -> Maybe TxKey) -> [txid] -> [TxKey]
forall a b. (a -> Maybe b) -> [a] -> [b]
mapMaybe (txid -> Map txid TxKey -> Maybe TxKey
forall k a. Ord k => k -> Map k a -> Maybe a
`Map.lookup` Map txid TxKey
submittedKeyOf) [txid]
acceptedTxIds
          rejectedKeys     = ((txid, err) -> Maybe TxKey) -> [(txid, err)] -> [TxKey]
forall a b. (a -> Maybe b) -> [a] -> [b]
mapMaybe ((txid -> Map txid TxKey -> Maybe TxKey
forall k a. Ord k => k -> Map k a -> Maybe a
`Map.lookup` Map txid TxKey
submittedKeyOf) (txid -> Maybe TxKey)
-> ((txid, err) -> txid) -> (txid, err) -> Maybe TxKey
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (txid, err) -> txid
forall a b. (a, b) -> a
fst) [(txid, err)]
rejectedTxs
          rejectedForTrace = ((txid, err) -> txid) -> [(txid, err)] -> [txid]
forall a b. (a -> b) -> [a] -> [b]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (txid, err) -> txid
forall a b. (a, b) -> a
fst [(txid, err)]
rejectedTxs
          acceptedCount    = [txid] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [txid]
acceptedTxIds
          rejectedCount    = [(txid, err)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(txid, err)]
rejectedTxs
          delta            = Time
end Time -> Time -> DiffTime
`diffTime` Time
start

      addCounters mempty { txSubmissionWaitMs = diffTimeToMilliseconds delta }
      peerState' <- applySubmittedTxs end resolvedTxKeys rejectedKeys peerState
      let (score, peerState'') =
            State.applyPeerEvents policy end acceptedCount rejectedCount peerState'
      traceWith tracer $
        TraceTxSubmissionProcessed ProcessedTxCount {
            ptxcAccepted = acceptedCount,
            ptxcRejected = rejectedCount,
            ptxcScore    = score
          }
      unless (null acceptedTxIds) $
        traceWith tracer (TraceTxInboundAddedToMempool acceptedTxIds delta)
      unless (null rejectedForTrace) $
        traceWith tracer (TraceTxInboundRejectedFromMempool rejectedForTrace delta)
      continueWithStateM k peerState''

    -- Request transaction bodies from the peer.
    requestTxBodies :: forall (n :: N).
                       Nat n
                    -> [TxKey]
                    -> StatefulM (PeerTxLocalState tx) n txid tx m
    requestTxBodies :: forall (n :: N).
Nat n -> [TxKey] -> StatefulM (PeerTxLocalState tx) n txid tx m
requestTxBodies Nat n
n [TxKey]
txKeys = (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> do
      txsToRequest <- PeerTxLocalState tx -> [TxKey] -> m (Map txid SizeInBytes)
resolveTxRequest PeerTxLocalState tx
peerState [TxKey]
txKeys
      traceWith tracer (TraceTxInboundRequestTxs (Map.keys txsToRequest))

      -- Record the start of the download episode on the first outstanding
      -- body request.  Subsequent pipelined requests leave the start time
      -- unchanged so we measure from first-send to last-receive.
      sendTime <- getMonotonicTime
      let peerState' = case PeerTxLocalState tx -> Maybe Time
forall tx. PeerTxLocalState tx -> Maybe Time
peerDownloadStartTime PeerTxLocalState tx
peerState of
                            Maybe Time
Nothing -> PeerTxLocalState tx
peerState { peerDownloadStartTime = Just sendTime }
                            Just Time
_  -> PeerTxLocalState tx
peerState
      pure $ SendMsgRequestTxsPipelined txsToRequest
               (continueWithStateM (continueAfterBodyRequests (Succ n)) peerState')

    -- Continue processing after receiving replies from the peer in pipelined mode.
    continueAfterReplies :: forall (n :: N).
                            Nat n
                         -> StatefulM (PeerTxLocalState tx) n txid tx m
    continueAfterReplies :: forall (n :: N).
Nat n -> StatefulM (PeerTxLocalState tx) n txid tx m
continueAfterReplies Nat n
Zero = StatefulM (PeerTxLocalState tx) n txid tx m
StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle
    continueAfterReplies n :: Nat n
n@Succ{} = (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> do
      now <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
      traceCanRequest n peerState
      (peerAction, peerState') <- runNextPeerActionPipelined now (State.drainPeerScore policy now peerState)
      case peerAction of
        PeerSubmitTxs [TxKey]
txKeys ->
          StatefulM (PeerTxLocalState tx) n txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle n txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM ([TxKey]
-> StatefulM (PeerTxLocalState tx) n txid tx m
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall (n :: N).
[TxKey]
-> StatefulM (PeerTxLocalState tx) n txid tx m
-> StatefulM (PeerTxLocalState tx) n txid tx m
submitBufferedTxs [TxKey]
txKeys (Nat n -> StatefulM (PeerTxLocalState tx) n txid tx m
forall (n :: N).
Nat n -> StatefulM (PeerTxLocalState tx) n txid tx m
continueAfterReplies Nat n
n)) PeerTxLocalState tx
peerState'
        PeerRequestTxs [TxKey]
txKeys ->
          StatefulM (PeerTxLocalState tx) n txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle n txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (Nat n -> [TxKey] -> StatefulM (PeerTxLocalState tx) n txid tx m
forall (n :: N).
Nat n -> [TxKey] -> StatefulM (PeerTxLocalState tx) n txid tx m
requestTxBodies Nat n
n [TxKey]
txKeys) PeerTxLocalState tx
peerState'
        PeerRequestTxIds NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq ->
          StatefulM (PeerTxLocalState tx) n txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle n txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (Nat n
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall (n :: N).
Nat n
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) n txid tx m
serverReqTxIds Nat n
n NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq) PeerTxLocalState tx
peerState'
        PeerDoNothing {} ->
          ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ()))
-> ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a b. (a -> b) -> a -> b
$ Stateful (PeerTxLocalState tx) n txid tx m
-> PeerTxLocalState tx -> ServerStIdle n txid tx m ()
forall s (n :: N) txid tx (m :: * -> *).
Stateful s n txid tx m -> s -> ServerStIdle n txid tx m ()
continueWithState (Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
handleReplies Nat n
Nat ('S n)
n) PeerTxLocalState tx
peerState'

    -- Continue processing after receiving transaction body replies in pipelined mode.
    continueAfterBodyRequests :: forall (n :: N).
                                 Nat (S n)
                              -> StatefulM (PeerTxLocalState tx) (S n) txid tx m
    continueAfterBodyRequests :: forall (n :: N).
Nat ('S n) -> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
continueAfterBodyRequests Nat ('S n)
n = (PeerTxLocalState tx -> m (ServerStIdle ('S n) txid tx m ()))
-> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle ('S n) txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) ('S n) txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle ('S n) txid tx m ()))
-> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> do
      now <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
      traceCanRequest n peerState
      (peerAction, peerState') <- runNextPeerActionPipelined now (State.drainPeerScore policy now peerState)
      case peerAction of
        PeerSubmitTxs [TxKey]
txKeys ->
          StatefulM (PeerTxLocalState tx) ('S n) txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle ('S n) txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM ([TxKey]
-> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
-> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
[TxKey]
-> StatefulM (PeerTxLocalState tx) n txid tx m
-> StatefulM (PeerTxLocalState tx) n txid tx m
submitBufferedTxs [TxKey]
txKeys (Nat ('S n) -> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat n -> StatefulM (PeerTxLocalState tx) n txid tx m
continueAfterReplies Nat ('S n)
n)) PeerTxLocalState tx
peerState'
        PeerRequestTxs [TxKey]
txKeys ->
          StatefulM (PeerTxLocalState tx) ('S n) txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle ('S n) txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (Nat ('S n)
-> [TxKey] -> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat n -> [TxKey] -> StatefulM (PeerTxLocalState tx) n txid tx m
requestTxBodies Nat ('S n)
n [TxKey]
txKeys) PeerTxLocalState tx
peerState'
        PeerRequestTxIds NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq ->
          StatefulM (PeerTxLocalState tx) ('S n) txid tx m
-> PeerTxLocalState tx -> m (ServerStIdle ('S n) txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulM s n txid tx m -> s -> m (ServerStIdle n txid tx m ())
continueWithStateM (Nat ('S n)
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat n
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) n txid tx m
serverReqTxIds Nat ('S n)
n NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq) PeerTxLocalState tx
peerState'
        PeerDoNothing {} ->
          ServerStIdle ('S n) txid tx m ()
-> m (ServerStIdle ('S n) txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle ('S n) txid tx m ()
 -> m (ServerStIdle ('S n) txid tx m ()))
-> ServerStIdle ('S n) txid tx m ()
-> m (ServerStIdle ('S n) txid tx m ())
forall a b. (a -> b) -> a -> b
$ Stateful (PeerTxLocalState tx) ('S n) txid tx m
-> PeerTxLocalState tx -> ServerStIdle ('S n) txid tx m ()
forall s (n :: N) txid tx (m :: * -> *).
Stateful s n txid tx m -> s -> ServerStIdle n txid tx m ()
continueWithState (Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
handleReplies Nat ('S n)
n) PeerTxLocalState tx
peerState'

    -- Construct and send a txid request message to the peer.
    serverReqTxIds :: forall (n :: N).
                      Nat n
                   -> NumTxIdsToAck
                   -> NumTxIdsToReq
                   -> StatefulM (PeerTxLocalState tx) n txid tx m
    -- No requests pending; transitions back to @serverIdle@
    serverReqTxIds :: forall (n :: N).
Nat n
-> NumTxIdsToAck
-> NumTxIdsToReq
-> StatefulM (PeerTxLocalState tx) n txid tx m
serverReqTxIds Nat n
Zero NumTxIdsToAck
0 NumTxIdsToReq
0 = StatefulM (PeerTxLocalState tx) n txid tx m
StatefulM (PeerTxLocalState tx) 'Z txid tx m
serverIdle

    -- Requests complete but pipeline not empty, continues to
    -- @handleReplies@ to process remaining in-flight replies
    serverReqTxIds n :: Nat n
n@Succ{} NumTxIdsToAck
0 NumTxIdsToReq
0 = (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState ->
      ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ()))
-> ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a b. (a -> b) -> a -> b
$ Stateful (PeerTxLocalState tx) n txid tx m
-> PeerTxLocalState tx -> ServerStIdle n txid tx m ()
forall s (n :: N) txid tx (m :: * -> *).
Stateful s n txid tx m -> s -> ServerStIdle n txid tx m ()
continueWithState (Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
handleReplies Nat n
Nat ('S n)
n) PeerTxLocalState tx
peerState

    -- Non-pipelined request, may send a blocking request
    serverReqTxIds Nat n
Zero NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq = (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState ->
      if StrictSeq TxKey -> Bool
forall a. StrictSeq a -> Bool
StrictSeq.null (PeerTxLocalState tx -> StrictSeq TxKey
forall tx. PeerTxLocalState tx -> StrictSeq TxKey
peerUnacknowledgedTxIds PeerTxLocalState tx
peerState)
         then do
           sendTime <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
           addCounters mempty { txIdBlockingReqsSent = 1 }
           pure $ SendMsgRequestTxIdsBlocking
                    txIdsToAck
                    txIdsToReq
                    (traceWith tracer TraceTxInboundTerminated)
                    (\NonEmpty (txid, SizeInBytes)
txids -> do
                        now <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
                        addCounters mempty { txIdBlockingWaitMs = diffTimeToMilliseconds (now `diffTime` sendTime) }
                        let txids' = NonEmpty (txid, SizeInBytes) -> [(txid, SizeInBytes)]
forall a. NonEmpty a -> [a]
NonEmpty.toList NonEmpty (txid, SizeInBytes)
txids
                        unless (length txids' <= fromIntegral txIdsToReq) $
                          throwIO ProtocolErrorTxIdsNotRequested
                        peerState' <- applyReceivedTxIds now txIdsToReq txids' peerState
                        continueWithStateM serverIdle peerState')
         else do
           TxSubmissionCounters -> m ()
addCounters TxSubmissionCounters
forall a. Monoid a => a
mempty { txIdPipelinedReqsSent = 1 }
           ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ()))
-> ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a b. (a -> b) -> a -> b
$ NumTxIdsToAck
-> NumTxIdsToReq
-> m (ServerStIdle ('S n) txid tx m ())
-> ServerStIdle n txid tx m ()
forall (m :: * -> *) (n :: N) txid tx a.
NumTxIdsToAck
-> NumTxIdsToReq
-> m (ServerStIdle ('S n) txid tx m a)
-> ServerStIdle n txid tx m a
SendMsgRequestTxIdsPipelined
                    NumTxIdsToAck
txIdsToAck
                    NumTxIdsToReq
txIdsToReq
                    (ServerStIdle ('S n) txid tx m ()
-> m (ServerStIdle ('S n) txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle ('S n) txid tx m ()
 -> m (ServerStIdle ('S n) txid tx m ()))
-> ServerStIdle ('S n) txid tx m ()
-> m (ServerStIdle ('S n) txid tx m ())
forall a b. (a -> b) -> a -> b
$ Stateful (PeerTxLocalState tx) ('S n) txid tx m
-> PeerTxLocalState tx -> ServerStIdle ('S n) txid tx m ()
forall s (n :: N) txid tx (m :: * -> *).
Stateful s n txid tx m -> s -> ServerStIdle n txid tx m ()
continueWithState (Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
handleReplies (Nat 'Z -> Nat ('S n)
forall (m :: N) (n :: N). (m ~ 'S n) => Nat n -> Nat m
Succ Nat 'Z
forall (n :: N). ('Z ~ n) => Nat n
Zero)) PeerTxLocalState tx
peerState)

    -- Pipelined request at depth > 0. Sends a pipelined message and continues
    -- to @handleReplies@.
    serverReqTxIds n :: Nat n
n@Succ{} NumTxIdsToAck
txIdsToAck NumTxIdsToReq
txIdsToReq = (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> m (ServerStIdle n txid tx m ())) -> StatefulM s n txid tx m
StatefulM ((PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulM (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx -> m (ServerStIdle n txid tx m ()))
-> StatefulM (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> do
      TxSubmissionCounters -> m ()
addCounters TxSubmissionCounters
forall a. Monoid a => a
mempty { txIdPipelinedReqsSent = 1 }
      ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ()))
-> ServerStIdle n txid tx m () -> m (ServerStIdle n txid tx m ())
forall a b. (a -> b) -> a -> b
$ NumTxIdsToAck
-> NumTxIdsToReq
-> m (ServerStIdle ('S n) txid tx m ())
-> ServerStIdle n txid tx m ()
forall (m :: * -> *) (n :: N) txid tx a.
NumTxIdsToAck
-> NumTxIdsToReq
-> m (ServerStIdle ('S n) txid tx m a)
-> ServerStIdle n txid tx m a
SendMsgRequestTxIdsPipelined
               NumTxIdsToAck
txIdsToAck
               NumTxIdsToReq
txIdsToReq
               (ServerStIdle ('S n) txid tx m ()
-> m (ServerStIdle ('S n) txid tx m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ServerStIdle ('S n) txid tx m ()
 -> m (ServerStIdle ('S n) txid tx m ()))
-> ServerStIdle ('S n) txid tx m ()
-> m (ServerStIdle ('S n) txid tx m ())
forall a b. (a -> b) -> a -> b
$ Stateful (PeerTxLocalState tx) ('S n) txid tx m
-> PeerTxLocalState tx -> ServerStIdle ('S n) txid tx m ()
forall s (n :: N) txid tx (m :: * -> *).
Stateful s n txid tx m -> s -> ServerStIdle n txid tx m ()
continueWithState (Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall (n :: N).
Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
handleReplies (Nat n -> Nat ('S n)
forall (m :: N) (n :: N). (m ~ 'S n) => Nat n -> Nat m
Succ Nat n
n)) PeerTxLocalState tx
peerState)

    -- Prepare to collect pipelined replies from the peer.
    handleReplies :: forall (n :: N).
                     Nat (S n)
                  -> Stateful (PeerTxLocalState tx) (S n) txid tx m
    handleReplies :: forall (n :: N).
Nat ('S n) -> Stateful (PeerTxLocalState tx) ('S n) txid tx m
handleReplies (Succ Nat n
n) = (PeerTxLocalState tx -> ServerStIdle ('S n) txid tx m ())
-> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> ServerStIdle n txid tx m ()) -> Stateful s n txid tx m
Stateful ((PeerTxLocalState tx -> ServerStIdle ('S n) txid tx m ())
 -> Stateful (PeerTxLocalState tx) ('S n) txid tx m)
-> (PeerTxLocalState tx -> ServerStIdle ('S n) txid tx m ())
-> Stateful (PeerTxLocalState tx) ('S n) txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState ->
      Maybe (m (ServerStIdle ('S n) txid tx m ()))
-> (Collect txid tx -> m (ServerStIdle n txid tx m ()))
-> ServerStIdle ('S n) txid tx m ()
forall (m :: * -> *) (n1 :: N) txid tx a.
Maybe (m (ServerStIdle ('S n1) txid tx m a))
-> (Collect txid tx -> m (ServerStIdle n1 txid tx m a))
-> ServerStIdle ('S n1) txid tx m a
CollectPipelined Maybe (m (ServerStIdle ('S n) txid tx m ()))
forall a. Maybe a
Nothing (StatefulCollect (PeerTxLocalState tx) n txid tx m
-> PeerTxLocalState tx
-> Collect txid tx
-> m (ServerStIdle n txid tx m ())
forall s (n :: N) txid tx (m :: * -> *).
StatefulCollect s n txid tx m
-> s -> Collect txid tx -> m (ServerStIdle n txid tx m ())
collectAndContinueWithState (Nat n -> StatefulCollect (PeerTxLocalState tx) n txid tx m
forall (n :: N).
Nat n -> StatefulCollect (PeerTxLocalState tx) n txid tx m
handleReply Nat n
Nat n
n) PeerTxLocalState tx
peerState)

    -- Process a single pipelined reply from the peer.
    handleReply :: forall (n :: N).
                   Nat n
                -> StatefulCollect (PeerTxLocalState tx) n txid tx m
    handleReply :: forall (n :: N).
Nat n -> StatefulCollect (PeerTxLocalState tx) n txid tx m
handleReply Nat n
n = (PeerTxLocalState tx
 -> Collect txid tx -> m (ServerStIdle n txid tx m ()))
-> StatefulCollect (PeerTxLocalState tx) n txid tx m
forall s (n :: N) txid tx (m :: * -> *).
(s -> Collect txid tx -> m (ServerStIdle n txid tx m ()))
-> StatefulCollect s n txid tx m
StatefulCollect ((PeerTxLocalState tx
  -> Collect txid tx -> m (ServerStIdle n txid tx m ()))
 -> StatefulCollect (PeerTxLocalState tx) n txid tx m)
-> (PeerTxLocalState tx
    -> Collect txid tx -> m (ServerStIdle n txid tx m ()))
-> StatefulCollect (PeerTxLocalState tx) n txid tx m
forall a b. (a -> b) -> a -> b
$ \PeerTxLocalState tx
peerState -> \case
      CollectTxIds NumTxIdsToReq
txIdsToReq [(txid, SizeInBytes)]
txids -> do
        Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([(txid, SizeInBytes)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(txid, SizeInBytes)]
txids Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= NumTxIdsToReq -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral NumTxIdsToReq
txIdsToReq) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
          TxSubmissionProtocolError -> m ()
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO TxSubmissionProtocolError
ProtocolErrorTxIdsNotRequested
        now <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        peerState' <- applyReceivedTxIds now txIdsToReq txids peerState
        continueWithStateM (continueAfterReplies n) peerState'

      CollectTxs Map txid SizeInBytes
requested [tx]
txs -> do
        let received :: Map txid tx
received = [(txid, tx)] -> Map txid tx
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList [ (tx -> txid
txId tx
tx, tx
tx) | tx
tx <- [tx]
txs ]
            wrongSizedTxs :: [(txid, SizeInBytes, SizeInBytes)]
wrongSizedTxs = Map txid SizeInBytes
-> Map txid tx -> [(txid, SizeInBytes, SizeInBytes)]
collectWrongSizedTxs Map txid SizeInBytes
requested Map txid tx
received
        Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Map txid tx -> Set txid
forall k a. Map k a -> Set k
Map.keysSet Map txid tx
received Set txid -> Set txid -> Bool
forall a. Ord a => Set a -> Set a -> Bool
`Set.isSubsetOf` Map txid SizeInBytes -> Set txid
forall k a. Map k a -> Set k
Map.keysSet Map txid SizeInBytes
requested) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
          TxSubmissionProtocolError -> m ()
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO TxSubmissionProtocolError
ProtocolErrorTxNotRequested
        Tracer m (TraceTxSubmissionInbound txid tx)
-> TraceTxSubmissionInbound txid tx -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceTxSubmissionInbound txid tx)
tracer (TraceTxSubmissionInbound txid tx -> m ())
-> TraceTxSubmissionInbound txid tx -> m ()
forall a b. (a -> b) -> a -> b
$ [txid] -> TraceTxSubmissionInbound txid tx
forall txid tx. [txid] -> TraceTxSubmissionInbound txid tx
TraceTxSubmissionCollected (tx -> txid
txId (tx -> txid) -> [tx] -> [txid]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [tx]
txs)
        Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([(txid, SizeInBytes, SizeInBytes)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(txid, SizeInBytes, SizeInBytes)]
wrongSizedTxs) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
          let protocolError :: TxSubmissionProtocolError
protocolError = [(txid, SizeInBytes, SizeInBytes)] -> TxSubmissionProtocolError
forall txid.
(Typeable txid, Show txid, Eq txid) =>
[(txid, SizeInBytes, SizeInBytes)] -> TxSubmissionProtocolError
ProtocolErrorTxSizeError [(txid, SizeInBytes, SizeInBytes)]
wrongSizedTxs
          Tracer m (TraceTxSubmissionInbound txid tx)
-> TraceTxSubmissionInbound txid tx -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceTxSubmissionInbound txid tx)
tracer (TxSubmissionProtocolError -> TraceTxSubmissionInbound txid tx
forall txid tx.
TxSubmissionProtocolError -> TraceTxSubmissionInbound txid tx
TraceTxInboundError TxSubmissionProtocolError
protocolError)
          TxSubmissionProtocolError -> m ()
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO TxSubmissionProtocolError
protocolError
        now <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        (penaltyCount, peerState') <- applyReceivedTxs now [ (txId tx, tx) | tx <- txs ] peerState
        peerState'' <-
          if penaltyCount == 0
             then pure peerState'
             else do
               let (score, ps) = State.applyPeerEvents policy now 0 penaltyCount peerState'
               traceWith tracer $
                 TraceTxSubmissionProcessed ProcessedTxCount {
                     ptxcAccepted = 0,
                     ptxcRejected = penaltyCount,
                     ptxcScore    = score
                   }
               pure ps
        continueWithStateM (continueAfterReplies n) peerState''

    -- Collect transactions with size mismatches between advertised and actual.
    collectWrongSizedTxs :: Map.Map txid SizeInBytes
                         -> Map.Map txid tx
                         -> [(txid, SizeInBytes, SizeInBytes)]
    collectWrongSizedTxs :: Map txid SizeInBytes
-> Map txid tx -> [(txid, SizeInBytes, SizeInBytes)]
collectWrongSizedTxs Map txid SizeInBytes
requestedTxIds Map txid tx
receivedTxs =
      [ (txid
txid', SizeInBytes
receivedSize, SizeInBytes
advertisedSize)
      | (txid
txid', tx
tx) <- Map txid tx -> [(txid, tx)]
forall k a. Map k a -> [(k, a)]
Map.toList Map txid tx
receivedTxs
      , let receivedSize :: SizeInBytes
receivedSize = tx -> SizeInBytes
txSize tx
tx
      , Just SizeInBytes
advertisedSize <- [txid -> Map txid SizeInBytes -> Maybe SizeInBytes
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup txid
txid' Map txid SizeInBytes
requestedTxIds]
      , Bool -> Bool
not (SizeInBytes -> SizeInBytes -> Bool
checkTxSize SizeInBytes
receivedSize SizeInBytes
advertisedSize)
      ]

    -- Fuzzy size comparison that allows for +/- const_MAX_TX_SIZE_DISCREPANCY.
    checkTxSize :: SizeInBytes
                -> SizeInBytes
                -> Bool
    checkTxSize :: SizeInBytes -> SizeInBytes -> Bool
checkTxSize SizeInBytes
received SizeInBytes
advertised
      | SizeInBytes
received SizeInBytes -> SizeInBytes -> Bool
forall a. Ord a => a -> a -> Bool
> SizeInBytes
advertised =
          SizeInBytes
received SizeInBytes -> SizeInBytes -> SizeInBytes
forall a. Num a => a -> a -> a
- SizeInBytes
advertised SizeInBytes -> SizeInBytes -> Bool
forall a. Ord a => a -> a -> Bool
<= SizeInBytes
const_MAX_TX_SIZE_DISCREPANCY
      | Bool
otherwise =
          SizeInBytes
advertised SizeInBytes -> SizeInBytes -> SizeInBytes
forall a. Num a => a -> a -> a
- SizeInBytes
received SizeInBytes -> SizeInBytes -> Bool
forall a. Ord a => a -> a -> Bool
<= SizeInBytes
const_MAX_TX_SIZE_DISCREPANCY