{-# LANGUAGE BangPatterns     #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE NamedFieldPuns   #-}
{-# LANGUAGE RecordWildCards  #-}
{-# LANGUAGE TypeFamilies     #-}
{-# LANGUAGE TypeOperators    #-}

module Ouroboros.Network.BlockFetch.State
  ( fetchLogicIterations
  , FetchDecisionPolicy (..)
  , FetchTriggerVariables (..)
  , FetchNonTriggerVariables (..)
  , FetchDecision
  , FetchDecline (..)
  , FetchMode (..)
  , PraosFetchMode (..)
  , TraceLabelPeer (..)
  , TraceFetchClientState (..)
  ) where

import Data.Functor.Contravariant (contramap)
import Data.Hashable (Hashable)
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Sequence (Seq (Empty))
import Data.Set qualified as Set
import Data.Void

import Control.Concurrent.Class.MonadSTM qualified as LazySTM
import Control.Concurrent.Class.MonadSTM.Strict.TVar.Checked (StrictTVar,
           newTVarIO, readTVarIO, writeTVar)
import Control.Exception (assert)
import Control.Monad.Class.MonadSTM
import Control.Monad.Class.MonadTime.SI
import Control.Monad.Class.MonadTimer.SI
import Control.Tracer (Tracer, traceWith)

import Ouroboros.Network.AnchoredFragment (AnchoredFragment)
import Ouroboros.Network.AnchoredFragment qualified as AF
import Ouroboros.Network.Block

import Ouroboros.Network.BlockFetch.ClientState (FetchClientStateVars (..),
           FetchRequest (..), PeerFetchInFlight (..), PeerFetchStatus (..),
           PeersOrder (..), TraceFetchClientState (..), TraceLabelPeer (..),
           addNewFetchRequest, readFetchClientState)
import Ouroboros.Network.BlockFetch.ConsensusInterface (ChainSelStarvation,
           FetchMode (..))
import Ouroboros.Network.BlockFetch.Decision (FetchDecision,
           FetchDecisionPolicy (..), FetchDecline (..), PeerInfo,
           PraosFetchMode (..), fetchDecisions)
import Ouroboros.Network.BlockFetch.Decision.Genesis (fetchDecisionsGenesisM)
import Ouroboros.Network.BlockFetch.Decision.Trace
import Ouroboros.Network.BlockFetch.DeltaQ (PeerGSV (..))

fetchLogicIterations
  :: ( HasHeader header
     , HasHeader block
     , HeaderHash header ~ HeaderHash block
     , MonadDelay m
     , MonadTimer m
     , Ord peer
     , Hashable peer
     )
  => Tracer m (TraceDecisionEvent peer header)
  -> Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
  -> FetchDecisionPolicy header
  -> FetchTriggerVariables peer header m
  -> FetchNonTriggerVariables peer header block m
  -> (peer -> m ()) -- ^ Action to call to demote the dynamo of ChainSync jumping.
  -> m Void
fetchLogicIterations :: forall header block (m :: * -> *) peer.
(HasHeader header, HasHeader block,
 HeaderHash header ~ HeaderHash block, MonadDelay m, MonadTimer m,
 Ord peer, Hashable peer) =>
Tracer m (TraceDecisionEvent peer header)
-> Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
-> FetchDecisionPolicy header
-> FetchTriggerVariables peer header m
-> FetchNonTriggerVariables peer header block m
-> (peer -> m ())
-> m Void
fetchLogicIterations Tracer m (TraceDecisionEvent peer header)
decisionTracer Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
clientStateTracer
                     FetchDecisionPolicy header
fetchDecisionPolicy
                     FetchTriggerVariables peer header m
fetchTriggerVariables
                     FetchNonTriggerVariables peer header block m
fetchNonTriggerVariables
                     peer -> m ()
demoteCSJDynamo = do

    peersOrderVar <- PeersOrder peer -> m (StrictTVar m (PeersOrder peer))
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO (PeersOrder peer -> m (StrictTVar m (PeersOrder peer)))
-> PeersOrder peer -> m (StrictTVar m (PeersOrder peer))
forall a b. (a -> b) -> a -> b
$ PeersOrder {
        peersOrderCurrent :: Maybe peer
peersOrderCurrent = Maybe peer
forall a. Maybe a
Nothing,
        peersOrderStart :: Time
peersOrderStart   = DiffTime -> Time
Time DiffTime
0,
        peersOrderAll :: Seq peer
peersOrderAll     = Seq peer
forall a. Seq a
Empty
      }

    iterateForever initialFetchStateFingerprint $ \FetchStateFingerprint peer header block
stateFingerprint -> do

      -- Run a single iteration of the fetch logic:
      --
      -- + wait for the state to change and make decisions for the new state
      -- + act on those decisions
      start <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
      (stateFingerprint', fetchMode) <- fetchLogicIteration
        decisionTracer clientStateTracer
        fetchDecisionPolicy
        fetchTriggerVariables
        fetchNonTriggerVariables
        stateFingerprint
        peersOrderVar
        demoteCSJDynamo
      end <- getMonotonicTime
      let delta = Time -> Time -> DiffTime
diffTime Time
end Time
start
          loopInterval = case FetchMode
fetchMode of
            FetchMode
FetchModeGenesis -> FetchDecisionPolicy header -> DiffTime
forall header. FetchDecisionPolicy header -> DiffTime
decisionLoopIntervalGenesis FetchDecisionPolicy header
fetchDecisionPolicy
            PraosFetchMode{} -> FetchDecisionPolicy header -> DiffTime
forall header. FetchDecisionPolicy header -> DiffTime
decisionLoopIntervalPraos FetchDecisionPolicy header
fetchDecisionPolicy
      -- Limit decision is made once every decisionLoopInterval.
      threadDelay (loopInterval - delta)
      pure stateFingerprint'


iterateForever :: Monad m => a -> (a -> m a) -> m Void
iterateForever :: forall (m :: * -> *) a. Monad m => a -> (a -> m a) -> m Void
iterateForever a
x0 a -> m a
m = a -> m Void
go a
x0 where go :: a -> m Void
go a
x = a -> m a
m a
x m a -> (a -> m Void) -> m Void
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= a -> m Void
go


-- | A single iteration of the fetch logic.
--
-- This involves:
--
-- * waiting for the state that the fetch decisions depend upon to change;
-- * taking a snapshot of the state;
-- * deciding for each peer if we will initiate a new fetch request
--
fetchLogicIteration
  :: (Hashable peer, Ord peer,
      HasHeader header, HasHeader block,
      HeaderHash header ~ HeaderHash block,
      MonadTimer m)
  => Tracer m (TraceDecisionEvent peer header)
  -> Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
  -> FetchDecisionPolicy header
  -> FetchTriggerVariables peer header m
  -> FetchNonTriggerVariables peer header block m
  -> FetchStateFingerprint peer header block
  -> StrictTVar m (PeersOrder peer)
  -> (peer -> m ()) -- ^ Action to call to demote the dynamo of ChainSync jumping.
  -> m (FetchStateFingerprint peer header block, FetchMode)
fetchLogicIteration :: forall peer header block (m :: * -> *).
(Hashable peer, Ord peer, HasHeader header, HasHeader block,
 HeaderHash header ~ HeaderHash block, MonadTimer m) =>
Tracer m (TraceDecisionEvent peer header)
-> Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
-> FetchDecisionPolicy header
-> FetchTriggerVariables peer header m
-> FetchNonTriggerVariables peer header block m
-> FetchStateFingerprint peer header block
-> StrictTVar m (PeersOrder peer)
-> (peer -> m ())
-> m (FetchStateFingerprint peer header block, FetchMode)
fetchLogicIteration Tracer m (TraceDecisionEvent peer header)
decisionTracer Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
clientStateTracer
                    FetchDecisionPolicy header
fetchDecisionPolicy
                    FetchTriggerVariables peer header m
fetchTriggerVariables
                    FetchNonTriggerVariables peer header block m
fetchNonTriggerVariables
                    FetchStateFingerprint peer header block
stateFingerprint
                    StrictTVar m (PeersOrder peer)
peersOrderVar
                    peer -> m ()
demoteCSJDynamo = do

    -- Gather a snapshot of all the state we need.
    --
    -- The grace period is considered to retrigger the decision logic even
    -- if no state has changed. This can help downloading blocks from a
    -- different peer if all ChainSync clients are blocked on the forecast
    -- horizon and the current peer of BlockFetch is not sending blocks.
    gracePeriodTVar <- DiffTime -> m (TVar m Bool)
forall (m :: * -> *). MonadTimer m => DiffTime -> m (TVar m Bool)
registerDelay (FetchDecisionPolicy header -> DiffTime
forall header. FetchDecisionPolicy header -> DiffTime
bulkSyncGracePeriod FetchDecisionPolicy header
fetchDecisionPolicy)
    (stateSnapshot, gracePeriodExpired, stateFingerprint') <-
      atomically $
        readStateVariables
          fetchTriggerVariables
          fetchNonTriggerVariables
          gracePeriodTVar
          stateFingerprint
    peersOrder <- readTVarIO peersOrderVar

    -- TODO: allow for boring PeerFetchStatusBusy transitions where we go round
    -- again rather than re-evaluating everything.
    assert (gracePeriodExpired || stateFingerprint' /= stateFingerprint) $ return ()

    -- TODO: log the difference in the fingerprint that caused us to wake up

    -- Make all the fetch decisions
    decisions <- fetchDecisionsForStateSnapshot
                      decisionTracer
                      fetchDecisionPolicy
                      stateSnapshot
                      (peersOrder,
                       atomically . writeTVar peersOrderVar,
                       demoteCSJDynamo)

    -- If we want to trace timings, we can do it here after forcing:
    -- _ <- evaluate (force decisions)

    -- Trace the batch of fetch decisions
    traceWith decisionTracer $ PeersFetch
      [ TraceLabelPeer peer (fmap fetchRequestPoints decision)
      | (decision, (_, _, _, peer, _)) <- decisions ]

    -- Tell the fetch clients to act on our decisions
    statusUpdates <- fetchLogicIterationAct clientStateTracer
                                            fetchDecisionPolicy
                                            (map swizzleReqVar decisions)
    let !stateFingerprint'' =
          [(peer, PeerFetchStatus header)]
-> FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block
forall peer header block.
Ord peer =>
[(peer, PeerFetchStatus header)]
-> FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block
updateFetchStateFingerprintPeerStatus [(peer, PeerFetchStatus header)]
statusUpdates FetchStateFingerprint peer header block
stateFingerprint'

    return (stateFingerprint'', fetchStateFetchMode stateSnapshot)
  where
    swizzleReqVar :: (a, (a, b, b, d, (c, d))) -> (a, b, c, d)
swizzleReqVar (a
d,(a
_,b
_,b
g,d
_,(c
rq,d
p))) = (a
d,b
g,c
rq,d
p)

    fetchRequestPoints :: HasHeader hdr => FetchRequest hdr -> [Point hdr]
    fetchRequestPoints :: forall hdr. HasHeader hdr => FetchRequest hdr -> [Point hdr]
fetchRequestPoints (FetchRequest [AnchoredFragment hdr]
headerss) =
      -- Flatten multiple fragments and trace points, not full headers
      [ hdr -> Point hdr
forall block. HasHeader block => block -> Point block
blockPoint hdr
header
      | AnchoredFragment hdr
headers <- [AnchoredFragment hdr]
headerss
      , hdr
header  <- AnchoredFragment hdr -> [hdr]
forall v a b. AnchoredSeq v a b -> [b]
AF.toOldestFirst AnchoredFragment hdr
headers ]

-- | Do a bit of rearranging of data before calling 'fetchDecisions' to do the
-- real work.
--
fetchDecisionsForStateSnapshot
  :: (HasHeader header,
      HeaderHash header ~ HeaderHash block,
      Ord peer,
      Hashable peer,
      MonadMonotonicTime m)
  => Tracer m (TraceDecisionEvent peer header)
  -> FetchDecisionPolicy header
  -> FetchStateSnapshot peer header block m
  -> ( PeersOrder peer
     , PeersOrder peer -> m ()
     , peer -> m ()
     )
  -> m [( FetchDecision (FetchRequest header),
          PeerInfo header peer (FetchClientStateVars m header, peer)
        )]

fetchDecisionsForStateSnapshot :: forall header block peer (m :: * -> *).
(HasHeader header, HeaderHash header ~ HeaderHash block, Ord peer,
 Hashable peer, MonadMonotonicTime m) =>
Tracer m (TraceDecisionEvent peer header)
-> FetchDecisionPolicy header
-> FetchStateSnapshot peer header block m
-> (PeersOrder peer, PeersOrder peer -> m (), peer -> m ())
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
fetchDecisionsForStateSnapshot
    Tracer m (TraceDecisionEvent peer header)
tracer
    FetchDecisionPolicy header
fetchDecisionPolicy
    FetchStateSnapshot {
      AnchoredFragment header
fetchStateCurrentChain :: AnchoredFragment header
fetchStateCurrentChain :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> AnchoredFragment header
fetchStateCurrentChain,
      Map peer (AnchoredFragment header)
fetchStatePeerChains :: Map peer (AnchoredFragment header)
fetchStatePeerChains :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m
-> Map peer (AnchoredFragment header)
fetchStatePeerChains,
      Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates :: Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m
-> Map
     peer
     (PeerFetchStatus header, PeerFetchInFlight header,
      FetchClientStateVars m header)
fetchStatePeerStates,
      Map peer PeerGSV
fetchStatePeerGSVs :: Map peer PeerGSV
fetchStatePeerGSVs :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> Map peer PeerGSV
fetchStatePeerGSVs,
      Point block -> Bool
fetchStateFetchedBlocks :: Point block -> Bool
fetchStateFetchedBlocks :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> Point block -> Bool
fetchStateFetchedBlocks,
      MaxSlotNo
fetchStateFetchedMaxSlotNo :: MaxSlotNo
fetchStateFetchedMaxSlotNo :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> MaxSlotNo
fetchStateFetchedMaxSlotNo,
      FetchMode
fetchStateFetchMode :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> FetchMode
fetchStateFetchMode :: FetchMode
fetchStateFetchMode,
      ChainSelStarvation
fetchStateChainSelStarvation :: ChainSelStarvation
fetchStateChainSelStarvation :: forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> ChainSelStarvation
fetchStateChainSelStarvation
    }
    (PeersOrder peer, PeersOrder peer -> m (), peer -> m ())
peersOrderHandlers =
    Bool
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a. HasCallStack => Bool -> a -> a
assert (                 Map peer (AnchoredFragment header) -> Set peer
forall k a. Map k a -> Set k
Map.keysSet Map peer (AnchoredFragment header)
fetchStatePeerChains
            Set peer -> Set peer -> Bool
forall a. Ord a => Set a -> Set a -> Bool
`Set.isSubsetOf` Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
-> Set peer
forall k a. Map k a -> Set k
Map.keysSet Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates) (m [(FetchDecision (FetchRequest header),
     PeerInfo header peer (FetchClientStateVars m header, peer))]
 -> m [(FetchDecision (FetchRequest header),
        PeerInfo header peer (FetchClientStateVars m header, peer))])
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a b. (a -> b) -> a -> b
$

    Bool
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a. HasCallStack => Bool -> a -> a
assert (                 Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
-> Set peer
forall k a. Map k a -> Set k
Map.keysSet Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates
            Set peer -> Set peer -> Bool
forall a. Ord a => Set a -> Set a -> Bool
`Set.isSubsetOf` Map peer PeerGSV -> Set peer
forall k a. Map k a -> Set k
Map.keysSet Map peer PeerGSV
fetchStatePeerGSVs) (m [(FetchDecision (FetchRequest header),
     PeerInfo header peer (FetchClientStateVars m header, peer))]
 -> m [(FetchDecision (FetchRequest header),
        PeerInfo header peer (FetchClientStateVars m header, peer))])
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a b. (a -> b) -> a -> b
$

    case FetchMode
fetchStateFetchMode of
      PraosFetchMode PraosFetchMode
fetchMode ->
        [(FetchDecision (FetchRequest header),
  PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([(FetchDecision (FetchRequest header),
   PeerInfo header peer (FetchClientStateVars m header, peer))]
 -> m [(FetchDecision (FetchRequest header),
        PeerInfo header peer (FetchClientStateVars m header, peer))])
-> [(FetchDecision (FetchRequest header),
     PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a b. (a -> b) -> a -> b
$ FetchDecisionPolicy header
-> PraosFetchMode
-> AnchoredFragment header
-> (Point block -> Bool)
-> MaxSlotNo
-> [(AnchoredFragment header,
     PeerInfo header peer (FetchClientStateVars m header, peer))]
-> [(FetchDecision (FetchRequest header),
     PeerInfo header peer (FetchClientStateVars m header, peer))]
forall peer header block extra.
(Ord peer, Hashable peer, HasHeader header,
 HeaderHash header ~ HeaderHash block) =>
FetchDecisionPolicy header
-> PraosFetchMode
-> AnchoredFragment header
-> (Point block -> Bool)
-> MaxSlotNo
-> [(AnchoredFragment header, PeerInfo header peer extra)]
-> [(FetchDecision (FetchRequest header),
     PeerInfo header peer extra)]
fetchDecisions
          FetchDecisionPolicy header
fetchDecisionPolicy
          PraosFetchMode
fetchMode
          AnchoredFragment header
fetchStateCurrentChain
          Point block -> Bool
fetchStateFetchedBlocks
          MaxSlotNo
fetchStateFetchedMaxSlotNo
          [(AnchoredFragment header,
  PeerInfo header peer (FetchClientStateVars m header, peer))]
peerChainsAndPeerInfo
      FetchMode
FetchModeGenesis ->
        Tracer m (TraceDecisionEvent peer header)
-> FetchDecisionPolicy header
-> AnchoredFragment header
-> (Point block -> Bool)
-> MaxSlotNo
-> ChainSelStarvation
-> (PeersOrder peer, PeersOrder peer -> m (), peer -> m ())
-> [(AnchoredFragment header,
     PeerInfo header peer (FetchClientStateVars m header, peer))]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer (FetchClientStateVars m header, peer))]
forall peer header block (m :: * -> *) extra.
(Ord peer, HasHeader header, HeaderHash header ~ HeaderHash block,
 MonadMonotonicTime m) =>
Tracer m (TraceDecisionEvent peer header)
-> FetchDecisionPolicy header
-> AnchoredFragment header
-> (Point block -> Bool)
-> MaxSlotNo
-> ChainSelStarvation
-> (PeersOrder peer, PeersOrder peer -> m (), peer -> m ())
-> [(AnchoredFragment header, PeerInfo header peer extra)]
-> m [(FetchDecision (FetchRequest header),
       PeerInfo header peer extra)]
fetchDecisionsGenesisM
          Tracer m (TraceDecisionEvent peer header)
tracer
          FetchDecisionPolicy header
fetchDecisionPolicy
          AnchoredFragment header
fetchStateCurrentChain
          Point block -> Bool
fetchStateFetchedBlocks
          MaxSlotNo
fetchStateFetchedMaxSlotNo
          ChainSelStarvation
fetchStateChainSelStarvation
          (PeersOrder peer, PeersOrder peer -> m (), peer -> m ())
peersOrderHandlers
          [(AnchoredFragment header,
  PeerInfo header peer (FetchClientStateVars m header, peer))]
peerChainsAndPeerInfo
  where
    peerChainsAndPeerInfo :: [(AnchoredFragment header,
  PeerInfo header peer (FetchClientStateVars m header, peer))]
peerChainsAndPeerInfo =
      ((peer,
  ((AnchoredFragment header,
    (PeerFetchStatus header, PeerFetchInFlight header,
     FetchClientStateVars m header)),
   PeerGSV))
 -> (AnchoredFragment header,
     PeerInfo header peer (FetchClientStateVars m header, peer)))
-> [(peer,
     ((AnchoredFragment header,
       (PeerFetchStatus header, PeerFetchInFlight header,
        FetchClientStateVars m header)),
      PeerGSV))]
-> [(AnchoredFragment header,
     PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a b. (a -> b) -> [a] -> [b]
map (peer,
 ((AnchoredFragment header,
   (PeerFetchStatus header, PeerFetchInFlight header,
    FetchClientStateVars m header)),
  PeerGSV))
-> (AnchoredFragment header,
    PeerInfo header peer (FetchClientStateVars m header, peer))
forall {b} {a} {a} {b} {a} {c}.
(b, ((a, (a, b, a)), c)) -> (a, (a, b, c, b, (a, b)))
swizzle ([(peer,
   ((AnchoredFragment header,
     (PeerFetchStatus header, PeerFetchInFlight header,
      FetchClientStateVars m header)),
    PeerGSV))]
 -> [(AnchoredFragment header,
      PeerInfo header peer (FetchClientStateVars m header, peer))])
-> (Map
      peer
      ((AnchoredFragment header,
        (PeerFetchStatus header, PeerFetchInFlight header,
         FetchClientStateVars m header)),
       PeerGSV)
    -> [(peer,
         ((AnchoredFragment header,
           (PeerFetchStatus header, PeerFetchInFlight header,
            FetchClientStateVars m header)),
          PeerGSV))])
-> Map
     peer
     ((AnchoredFragment header,
       (PeerFetchStatus header, PeerFetchInFlight header,
        FetchClientStateVars m header)),
      PeerGSV)
-> [(AnchoredFragment header,
     PeerInfo header peer (FetchClientStateVars m header, peer))]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Map
  peer
  ((AnchoredFragment header,
    (PeerFetchStatus header, PeerFetchInFlight header,
     FetchClientStateVars m header)),
   PeerGSV)
-> [(peer,
     ((AnchoredFragment header,
       (PeerFetchStatus header, PeerFetchInFlight header,
        FetchClientStateVars m header)),
      PeerGSV))]
forall k a. Map k a -> [(k, a)]
Map.toList (Map
   peer
   ((AnchoredFragment header,
     (PeerFetchStatus header, PeerFetchInFlight header,
      FetchClientStateVars m header)),
    PeerGSV)
 -> [(AnchoredFragment header,
      PeerInfo header peer (FetchClientStateVars m header, peer))])
-> Map
     peer
     ((AnchoredFragment header,
       (PeerFetchStatus header, PeerFetchInFlight header,
        FetchClientStateVars m header)),
      PeerGSV)
-> [(AnchoredFragment header,
     PeerInfo header peer (FetchClientStateVars m header, peer))]
forall a b. (a -> b) -> a -> b
$
      ((AnchoredFragment header,
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header))
 -> PeerGSV
 -> ((AnchoredFragment header,
      (PeerFetchStatus header, PeerFetchInFlight header,
       FetchClientStateVars m header)),
     PeerGSV))
-> Map
     peer
     (AnchoredFragment header,
      (PeerFetchStatus header, PeerFetchInFlight header,
       FetchClientStateVars m header))
-> Map peer PeerGSV
-> Map
     peer
     ((AnchoredFragment header,
       (PeerFetchStatus header, PeerFetchInFlight header,
        FetchClientStateVars m header)),
      PeerGSV)
forall k a b c.
Ord k =>
(a -> b -> c) -> Map k a -> Map k b -> Map k c
Map.intersectionWith (,)
        ((AnchoredFragment header
 -> (PeerFetchStatus header, PeerFetchInFlight header,
     FetchClientStateVars m header)
 -> (AnchoredFragment header,
     (PeerFetchStatus header, PeerFetchInFlight header,
      FetchClientStateVars m header)))
-> Map peer (AnchoredFragment header)
-> Map
     peer
     (PeerFetchStatus header, PeerFetchInFlight header,
      FetchClientStateVars m header)
-> Map
     peer
     (AnchoredFragment header,
      (PeerFetchStatus header, PeerFetchInFlight header,
       FetchClientStateVars m header))
forall k a b c.
Ord k =>
(a -> b -> c) -> Map k a -> Map k b -> Map k c
Map.intersectionWith (,) Map peer (AnchoredFragment header)
fetchStatePeerChains Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates)
        Map peer PeerGSV
fetchStatePeerGSVs

    swizzle :: (b, ((a, (a, b, a)), c)) -> (a, (a, b, c, b, (a, b)))
swizzle (b
peer, ((a
chain, (a
status, b
inflight, a
vars)), c
gsvs)) =
      (a
chain, (a
status, b
inflight, c
gsvs, b
peer, (a
vars, b
peer)))


-- | Act on decisions to send new requests. In fact all we do here is update
-- request variables that are shared with the threads running the block fetch
-- protocol with each peer.
--
fetchLogicIterationAct :: (MonadSTM m, HasHeader header)
                       => Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
                       -> FetchDecisionPolicy header
                       -> [(FetchDecision (FetchRequest header),
                            PeerGSV,
                            FetchClientStateVars m header,
                            peer)]
                       -> m [(peer, PeerFetchStatus header)]
fetchLogicIterationAct :: forall (m :: * -> *) header peer.
(MonadSTM m, HasHeader header) =>
Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
-> FetchDecisionPolicy header
-> [(FetchDecision (FetchRequest header), PeerGSV,
     FetchClientStateVars m header, peer)]
-> m [(peer, PeerFetchStatus header)]
fetchLogicIterationAct Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
clientStateTracer FetchDecisionPolicy{header -> SizeInBytes
blockFetchSize :: header -> SizeInBytes
blockFetchSize :: forall header. FetchDecisionPolicy header -> header -> SizeInBytes
blockFetchSize}
                       [(FetchDecision (FetchRequest header), PeerGSV,
  FetchClientStateVars m header, peer)]
decisions =
    [m (peer, PeerFetchStatus header)]
-> m [(peer, PeerFetchStatus header)]
forall (t :: * -> *) (m :: * -> *) a.
(Traversable t, Monad m) =>
t (m a) -> m (t a)
forall (m :: * -> *) a. Monad m => [m a] -> m [a]
sequence
      [ (,) peer
peer (PeerFetchStatus header -> (peer, PeerFetchStatus header))
-> m (PeerFetchStatus header) -> m (peer, PeerFetchStatus header)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Tracer m (TraceFetchClientState header)
-> (header -> SizeInBytes)
-> FetchRequest header
-> PeerGSV
-> FetchClientStateVars m header
-> m (PeerFetchStatus header)
forall (m :: * -> *) header.
(MonadSTM m, HasHeader header) =>
Tracer m (TraceFetchClientState header)
-> (header -> SizeInBytes)
-> FetchRequest header
-> PeerGSV
-> FetchClientStateVars m header
-> m (PeerFetchStatus header)
addNewFetchRequest
                       ((TraceFetchClientState header
 -> TraceLabelPeer peer (TraceFetchClientState header))
-> Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
-> Tracer m (TraceFetchClientState header)
forall a' a. (a' -> a) -> Tracer m a -> Tracer m a'
forall (f :: * -> *) a' a.
Contravariant f =>
(a' -> a) -> f a -> f a'
contramap (peer
-> TraceFetchClientState header
-> TraceLabelPeer peer (TraceFetchClientState header)
forall peerid a. peerid -> a -> TraceLabelPeer peerid a
TraceLabelPeer peer
peer) Tracer m (TraceLabelPeer peer (TraceFetchClientState header))
clientStateTracer)
                       header -> SizeInBytes
blockFetchSize
                       FetchRequest header
request PeerGSV
gsvs
                       FetchClientStateVars m header
stateVars
      | (Right FetchRequest header
request, PeerGSV
gsvs, FetchClientStateVars m header
stateVars, peer
peer) <- [(FetchDecision (FetchRequest header), PeerGSV,
  FetchClientStateVars m header, peer)]
decisions ]


-- | STM actions to read various state variables that the fetch logic depends
-- upon. Any change in these variables is a trigger to re-evaluate the decision
-- on what blocks to fetch.
--
-- Note that this is a \"level trigger\" not an \"edge trigger\": we do not
-- have to re-evaluate on every change, it is sufficient to re-evaluate at some
-- stage after one or more changes. This means it is ok to get somewhat behind,
-- and it is not necessary to determine exactly what changed, just that there
-- was some change.
--
data FetchTriggerVariables peer header m = FetchTriggerVariables {
       forall peer header (m :: * -> *).
FetchTriggerVariables peer header m
-> STM m (AnchoredFragment header)
readStateCurrentChain    :: STM m (AnchoredFragment header),
       forall peer header (m :: * -> *).
FetchTriggerVariables peer header m
-> STM m (Map peer (AnchoredFragment header))
readStateCandidateChains :: STM m (Map peer (AnchoredFragment header)),
       forall peer header (m :: * -> *).
FetchTriggerVariables peer header m
-> STM m (Map peer (PeerFetchStatus header))
readStatePeerStatus      :: STM m (Map peer (PeerFetchStatus header))
     }

-- | STM actions to read various state variables that the fetch logic uses.
-- While the decisions do make use of the values of these variables, it is not
-- necessary to re-evaluate when these variables change.
--
data FetchNonTriggerVariables peer header block m = FetchNonTriggerVariables {
       forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m (Point block -> Bool)
readStateFetchedBlocks    :: STM m (Point block -> Bool),
       forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m (Map peer (FetchClientStateVars m header))
readStatePeerStateVars    :: STM m (Map peer (FetchClientStateVars m header)),
       forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m (Map peer PeerGSV)
readStatePeerGSVs         :: STM m (Map peer PeerGSV),
       forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m -> STM m FetchMode
readStateFetchMode        :: STM m FetchMode,
       forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m -> STM m MaxSlotNo
readStateFetchedMaxSlotNo :: STM m MaxSlotNo,
       forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m ChainSelStarvation
readStateChainSelStarvation :: STM m ChainSelStarvation
     }


data FetchStateFingerprint peer header block =
     FetchStateFingerprint
       !(Maybe (Point block))
       !(Map peer (Point header))
       !(Map peer (PeerFetchStatus header))
  deriving FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block -> Bool
(FetchStateFingerprint peer header block
 -> FetchStateFingerprint peer header block -> Bool)
-> (FetchStateFingerprint peer header block
    -> FetchStateFingerprint peer header block -> Bool)
-> Eq (FetchStateFingerprint peer header block)
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
forall peer header block.
(StandardHash block, StandardHash header, Eq peer) =>
FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block -> Bool
$c== :: forall peer header block.
(StandardHash block, StandardHash header, Eq peer) =>
FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block -> Bool
== :: FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block -> Bool
$c/= :: forall peer header block.
(StandardHash block, StandardHash header, Eq peer) =>
FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block -> Bool
/= :: FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block -> Bool
Eq

initialFetchStateFingerprint :: FetchStateFingerprint peer header block
initialFetchStateFingerprint :: forall peer header block. FetchStateFingerprint peer header block
initialFetchStateFingerprint =
    Maybe (Point block)
-> Map peer (Point header)
-> Map peer (PeerFetchStatus header)
-> FetchStateFingerprint peer header block
forall peer header block.
Maybe (Point block)
-> Map peer (Point header)
-> Map peer (PeerFetchStatus header)
-> FetchStateFingerprint peer header block
FetchStateFingerprint
      Maybe (Point block)
forall a. Maybe a
Nothing
      Map peer (Point header)
forall k a. Map k a
Map.empty
      Map peer (PeerFetchStatus header)
forall k a. Map k a
Map.empty

updateFetchStateFingerprintPeerStatus :: Ord peer
                                      => [(peer, PeerFetchStatus header)]
                                      -> FetchStateFingerprint peer header block
                                      -> FetchStateFingerprint peer header block
updateFetchStateFingerprintPeerStatus :: forall peer header block.
Ord peer =>
[(peer, PeerFetchStatus header)]
-> FetchStateFingerprint peer header block
-> FetchStateFingerprint peer header block
updateFetchStateFingerprintPeerStatus [(peer, PeerFetchStatus header)]
statuses'
    (FetchStateFingerprint Maybe (Point block)
current Map peer (Point header)
candidates Map peer (PeerFetchStatus header)
statuses) =
    Maybe (Point block)
-> Map peer (Point header)
-> Map peer (PeerFetchStatus header)
-> FetchStateFingerprint peer header block
forall peer header block.
Maybe (Point block)
-> Map peer (Point header)
-> Map peer (PeerFetchStatus header)
-> FetchStateFingerprint peer header block
FetchStateFingerprint
      Maybe (Point block)
current
      Map peer (Point header)
candidates
      (Map peer (PeerFetchStatus header)
-> Map peer (PeerFetchStatus header)
-> Map peer (PeerFetchStatus header)
forall k a. Ord k => Map k a -> Map k a -> Map k a
Map.union ([(peer, PeerFetchStatus header)]
-> Map peer (PeerFetchStatus header)
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList [(peer, PeerFetchStatus header)]
statuses') Map peer (PeerFetchStatus header)
statuses) -- left overrides right

-- |
--
-- Note that the domain of 'fetchStatePeerChains' is a subset of the domain
-- of 'fetchStatePeerStates' and 'fetchStatePeerReqVars'.
--
data FetchStateSnapshot peer header block m = FetchStateSnapshot {
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> AnchoredFragment header
fetchStateCurrentChain     :: AnchoredFragment header,
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m
-> Map peer (AnchoredFragment header)
fetchStatePeerChains       :: Map peer (AnchoredFragment header),
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m
-> Map
     peer
     (PeerFetchStatus header, PeerFetchInFlight header,
      FetchClientStateVars m header)
fetchStatePeerStates       :: Map peer (PeerFetchStatus   header,
                                               PeerFetchInFlight header,
                                               FetchClientStateVars m header),
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> Map peer PeerGSV
fetchStatePeerGSVs         :: Map peer PeerGSV,
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> Point block -> Bool
fetchStateFetchedBlocks    :: Point block -> Bool,
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> FetchMode
fetchStateFetchMode        :: FetchMode,
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> MaxSlotNo
fetchStateFetchedMaxSlotNo :: MaxSlotNo,
       forall peer header block (m :: * -> *).
FetchStateSnapshot peer header block m -> ChainSelStarvation
fetchStateChainSelStarvation :: ChainSelStarvation
     }

readStateVariables :: (MonadSTM m, Eq peer,
                       HasHeader header, HasHeader block,
                       HeaderHash header ~ HeaderHash block)
                   => FetchTriggerVariables peer header m
                   -> FetchNonTriggerVariables peer header block m
                   -> LazySTM.TVar m Bool
                   -> FetchStateFingerprint peer header block
                   -> STM m (FetchStateSnapshot peer header block m,
                             Bool,
                             FetchStateFingerprint peer header block)
readStateVariables :: forall (m :: * -> *) peer header block.
(MonadSTM m, Eq peer, HasHeader header, HasHeader block,
 HeaderHash header ~ HeaderHash block) =>
FetchTriggerVariables peer header m
-> FetchNonTriggerVariables peer header block m
-> TVar m Bool
-> FetchStateFingerprint peer header block
-> STM
     m
     (FetchStateSnapshot peer header block m, Bool,
      FetchStateFingerprint peer header block)
readStateVariables FetchTriggerVariables{STM m (Map peer (AnchoredFragment header))
STM m (Map peer (PeerFetchStatus header))
STM m (AnchoredFragment header)
readStateCurrentChain :: forall peer header (m :: * -> *).
FetchTriggerVariables peer header m
-> STM m (AnchoredFragment header)
readStateCandidateChains :: forall peer header (m :: * -> *).
FetchTriggerVariables peer header m
-> STM m (Map peer (AnchoredFragment header))
readStatePeerStatus :: forall peer header (m :: * -> *).
FetchTriggerVariables peer header m
-> STM m (Map peer (PeerFetchStatus header))
readStateCurrentChain :: STM m (AnchoredFragment header)
readStateCandidateChains :: STM m (Map peer (AnchoredFragment header))
readStatePeerStatus :: STM m (Map peer (PeerFetchStatus header))
..}
                   FetchNonTriggerVariables{STM m (Map peer PeerGSV)
STM m (Map peer (FetchClientStateVars m header))
STM m MaxSlotNo
STM m ChainSelStarvation
STM m FetchMode
STM m (Point block -> Bool)
readStateFetchedBlocks :: forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m (Point block -> Bool)
readStatePeerStateVars :: forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m (Map peer (FetchClientStateVars m header))
readStatePeerGSVs :: forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m (Map peer PeerGSV)
readStateFetchMode :: forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m -> STM m FetchMode
readStateFetchedMaxSlotNo :: forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m -> STM m MaxSlotNo
readStateChainSelStarvation :: forall peer header block (m :: * -> *).
FetchNonTriggerVariables peer header block m
-> STM m ChainSelStarvation
readStateFetchedBlocks :: STM m (Point block -> Bool)
readStatePeerStateVars :: STM m (Map peer (FetchClientStateVars m header))
readStatePeerGSVs :: STM m (Map peer PeerGSV)
readStateFetchMode :: STM m FetchMode
readStateFetchedMaxSlotNo :: STM m MaxSlotNo
readStateChainSelStarvation :: STM m ChainSelStarvation
..}
                   TVar m Bool
gracePeriodTVar
                   FetchStateFingerprint peer header block
fetchStateFingerprint = do

    -- Read all the trigger state variables
    fetchStateCurrentChain  <- STM m (AnchoredFragment header)
readStateCurrentChain
    fetchStatePeerChains    <- readStateCandidateChains
    fetchStatePeerStatus    <- readStatePeerStatus
    gracePeriodExpired      <- LazySTM.readTVar gracePeriodTVar

    -- Construct the change detection fingerprint
    let !fetchStateFingerprint' =
          Maybe (Point block)
-> Map peer (Point header)
-> Map peer (PeerFetchStatus header)
-> FetchStateFingerprint peer header block
forall peer header block.
Maybe (Point block)
-> Map peer (Point header)
-> Map peer (PeerFetchStatus header)
-> FetchStateFingerprint peer header block
FetchStateFingerprint
            (Point block -> Maybe (Point block)
forall a. a -> Maybe a
Just (Point header -> Point block
forall {k1} {k2} (b :: k1) (b' :: k2).
Coercible (HeaderHash b) (HeaderHash b') =>
Point b -> Point b'
castPoint (AnchoredFragment header -> Point header
forall block.
HasHeader block =>
AnchoredFragment block -> Point block
AF.headPoint AnchoredFragment header
fetchStateCurrentChain)))
            ((AnchoredFragment header -> Point header)
-> Map peer (AnchoredFragment header) -> Map peer (Point header)
forall a b k. (a -> b) -> Map k a -> Map k b
Map.map AnchoredFragment header -> Point header
forall block.
HasHeader block =>
AnchoredFragment block -> Point block
AF.headPoint Map peer (AnchoredFragment header)
fetchStatePeerChains)
            Map peer (PeerFetchStatus header)
fetchStatePeerStatus

    -- Check the fingerprint changed, or block and wait until it does
    check (gracePeriodExpired || fetchStateFingerprint' /= fetchStateFingerprint)

    -- Now read all the non-trigger state variables
    fetchStatePeerStates       <- readStatePeerStateVars
                              >>= traverse readFetchClientState
    fetchStatePeerGSVs         <- readStatePeerGSVs
    fetchStateFetchedBlocks    <- readStateFetchedBlocks
    fetchStateFetchMode        <- readStateFetchMode
    fetchStateFetchedMaxSlotNo <- readStateFetchedMaxSlotNo
    fetchStateChainSelStarvation <- readStateChainSelStarvation

    -- Construct the overall snapshot of the state
    let fetchStateSnapshot =
          FetchStateSnapshot {
            AnchoredFragment header
fetchStateCurrentChain :: AnchoredFragment header
fetchStateCurrentChain :: AnchoredFragment header
fetchStateCurrentChain,
            Map peer (AnchoredFragment header)
fetchStatePeerChains :: Map peer (AnchoredFragment header)
fetchStatePeerChains :: Map peer (AnchoredFragment header)
fetchStatePeerChains,
            Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates :: Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates :: Map
  peer
  (PeerFetchStatus header, PeerFetchInFlight header,
   FetchClientStateVars m header)
fetchStatePeerStates,
            Map peer PeerGSV
fetchStatePeerGSVs :: Map peer PeerGSV
fetchStatePeerGSVs :: Map peer PeerGSV
fetchStatePeerGSVs,
            Point block -> Bool
fetchStateFetchedBlocks :: Point block -> Bool
fetchStateFetchedBlocks :: Point block -> Bool
fetchStateFetchedBlocks,
            FetchMode
fetchStateFetchMode :: FetchMode
fetchStateFetchMode :: FetchMode
fetchStateFetchMode,
            MaxSlotNo
fetchStateFetchedMaxSlotNo :: MaxSlotNo
fetchStateFetchedMaxSlotNo :: MaxSlotNo
fetchStateFetchedMaxSlotNo,
            ChainSelStarvation
fetchStateChainSelStarvation :: ChainSelStarvation
fetchStateChainSelStarvation :: ChainSelStarvation
fetchStateChainSelStarvation
          }

    return (fetchStateSnapshot, gracePeriodExpired, fetchStateFingerprint')