{-# LANGUAGE BlockArguments #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}

-- | A test for the consensus-specific parts of the BlockFetch client.
--
-- When adding a block to the ChainDB, we allocate potential punishments, which
-- are later invoked after block validation, crucially allowing us to kill the
-- BlockFetch client and hence disconnect from malicious peers.
--
-- This test spins up several BlockFetch clients, which download randomly
-- generated chains and add them to the ChainDB, which will enact these
-- punishments on validation. Right now, we only ensure that doing so for chains
-- originating from honest behavior do not cause any disconnects, but we plan to
-- also model malicious/erroneous behavior.
module Test.Consensus.MiniProtocol.BlockFetch.Client (tests) where

import           Control.Monad (replicateM)
import           Control.Monad.Class.MonadTime
import           Control.Monad.Class.MonadTimer.SI (MonadTimer)
import           Control.Monad.IOSim (runSimOrThrow)
import           Control.ResourceRegistry
import           Control.Tracer (Tracer (..), nullTracer, traceWith)
import           Data.Bifunctor (first)
import           Data.Hashable (Hashable)
import           Data.Map.Strict (Map)
import qualified Data.Map.Strict as Map
import           Data.Traversable (for)
import           Network.TypedProtocol.Channel (createConnectedChannels)
import           Network.TypedProtocol.Codec (AnyMessage (..))
import           Network.TypedProtocol.Core (PeerRole (..))
import qualified Network.TypedProtocol.Driver.Simple as Driver
import           Ouroboros.Consensus.Block
import           Ouroboros.Consensus.Config
import qualified Ouroboros.Consensus.MiniProtocol.BlockFetch.ClientInterface as BlockFetchClientInterface
import           Ouroboros.Consensus.Node.ProtocolInfo (NumCoreNodes (..))
import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
import qualified Ouroboros.Consensus.Storage.ChainDB.Impl as ChainDBImpl
import qualified Ouroboros.Consensus.Storage.ChainDB.Impl.Args as ChainDB
import           Ouroboros.Consensus.Util.Condense (Condense (..))
import           Ouroboros.Consensus.Util.IOLike
import           Ouroboros.Consensus.Util.STM (blockUntilJust,
                     forkLinkedWatcher)
import           Ouroboros.Network.AnchoredFragment (AnchoredFragment)
import qualified Ouroboros.Network.AnchoredFragment as AF
import           Ouroboros.Network.BlockFetch (BlockFetchConfiguration (..),
                     BlockFetchConsensusInterface, FetchMode (..),
                     blockFetchLogic, bracketFetchClient,
                     bracketKeepAliveClient, bracketSyncWithFetchClient,
                     newFetchClientRegistry)
import           Ouroboros.Network.BlockFetch.Client (blockFetchClient)
import           Ouroboros.Network.ControlMessage (ControlMessage (..))
import           Ouroboros.Network.Mock.Chain (Chain)
import qualified Ouroboros.Network.Mock.Chain as Chain
import           Ouroboros.Network.NodeToNode.Version (NodeToNodeVersion)
import           Ouroboros.Network.Protocol.BlockFetch.Codec (codecBlockFetchId)
import           Ouroboros.Network.Protocol.BlockFetch.Server
                     (BlockFetchBlockSender (SendMsgNoBlocks, SendMsgStartBatch),
                     BlockFetchSendBlocks (SendMsgBatchDone, SendMsgBlock),
                     BlockFetchServer (..), blockFetchServerPeer)
import           Ouroboros.Network.Protocol.BlockFetch.Type (BlockFetch,
                     ChainRange (..), Message (MsgBlock))
import           Test.QuickCheck
import           Test.Tasty
import           Test.Tasty.QuickCheck
import           Test.Util.ChainDB
import           Test.Util.ChainUpdates
import qualified Test.Util.LogicalClock as LogicalClock
import           Test.Util.LogicalClock (Tick (..))
import           Test.Util.Orphans.IOLike ()
import           Test.Util.Schedule
import           Test.Util.TestBlock
import           Test.Util.Time (dawnOfTime)
import           Test.Util.Tracer (recordingTracerTVar)

tests :: TestTree
tests :: TestTree
tests = String -> [TestTree] -> TestTree
testGroup String
"BlockFetchClient"
    [ String -> (BlockFetchClientTestSetup -> Property) -> TestTree
forall a. Testable a => String -> a -> TestTree
testProperty String
"blockFetch" BlockFetchClientTestSetup -> Property
prop_blockFetch
    ]

prop_blockFetch :: BlockFetchClientTestSetup -> Property
prop_blockFetch :: BlockFetchClientTestSetup -> Property
prop_blockFetch bfcts :: BlockFetchClientTestSetup
bfcts@BlockFetchClientTestSetup{Map PeerId (Schedule ChainUpdate)
DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
peerUpdates :: Map PeerId (Schedule ChainUpdate)
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
peerUpdates :: BlockFetchClientTestSetup -> Map PeerId (Schedule ChainUpdate)
blockFetchMode :: BlockFetchClientTestSetup -> FetchMode
blockFetchCfg :: BlockFetchClientTestSetup -> BlockFetchConfiguration
blockFetchPipelining :: BlockFetchClientTestSetup -> DiffusionPipeliningSupport
..} =
    String -> Property -> Property
forall prop. Testable prop => String -> prop -> Property
counterexample (String
"Trace:\n" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> [String] -> String
unlines ((Tick, String) -> String
ppTrace ((Tick, String) -> String) -> [(Tick, String)] -> [String]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [(Tick, String)]
bfcoTrace)) (Property -> Property) -> Property -> Property
forall a b. (a -> b) -> a -> b
$
    String -> Property -> Property
forall prop. Testable prop => String -> prop -> Property
counterexample (BlockFetchClientTestSetup -> String
forall a. Condense a => a -> String
condense BlockFetchClientTestSetup
bfcts) (Property -> Property) -> Property -> Property
forall a b. (a -> b) -> a -> b
$
    [Property] -> Property
forall prop. Testable prop => [prop] -> Property
conjoin ([Property] -> Property) -> [Property] -> Property
forall a b. (a -> b) -> a -> b
$
      [ String -> Either SomeException () -> Property
forall {e}. Exception e => String -> Either e () -> Property
noException (String
"BlockFetch client " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> PeerId -> String
forall a. Condense a => a -> String
condense PeerId
peerId) Either SomeException ()
res
      | (PeerId
peerId, Either SomeException ()
res) <- Map PeerId (Either SomeException ())
-> [(PeerId, Either SomeException ())]
forall k a. Map k a -> [(k, a)]
Map.toList Map PeerId (Either SomeException ())
bfcoBlockFetchResults
      ] [Property] -> [Property] -> [Property]
forall a. Semigroup a => a -> a -> a
<>
      [ Map PeerId (Either SomeException ()) -> Set PeerId
forall k a. Map k a -> Set k
Map.keysSet Map PeerId (Either SomeException ())
bfcoBlockFetchResults Set PeerId -> Set PeerId -> Property
forall a. (Eq a, Show a) => a -> a -> Property
=== Map PeerId (Schedule ChainUpdate) -> Set PeerId
forall k a. Map k a -> Set k
Map.keysSet Map PeerId (Schedule ChainUpdate)
peerUpdates
      , String -> Property -> Property
forall prop. Testable prop => String -> prop -> Property
counterexample (String
"Fetched blocks per peer: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Map PeerId Word -> String
forall a. Condense a => a -> String
condense Map PeerId Word
bfcoFetchedBlocks) (Property -> Property) -> Property -> Property
forall a b. (a -> b) -> a -> b
$
        Bool -> Property
forall prop. Testable prop => prop -> Property
property (Bool -> Property) -> Bool -> Property
forall a b. (a -> b) -> a -> b
$ (Word -> Bool) -> Map PeerId Word -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
all (Word -> Word -> Bool
forall a. Ord a => a -> a -> Bool
> Word
0) Map PeerId Word
bfcoFetchedBlocks
      ]
  where
    BlockFetchClientOutcome{[(Tick, String)]
Map PeerId Word
Map PeerId (Either SomeException ())
bfcoTrace :: [(Tick, String)]
bfcoBlockFetchResults :: Map PeerId (Either SomeException ())
bfcoFetchedBlocks :: Map PeerId Word
bfcoBlockFetchResults :: BlockFetchClientOutcome -> Map PeerId (Either SomeException ())
bfcoFetchedBlocks :: BlockFetchClientOutcome -> Map PeerId Word
bfcoTrace :: BlockFetchClientOutcome -> [(Tick, String)]
..} = (forall s. IOSim s BlockFetchClientOutcome)
-> BlockFetchClientOutcome
forall a. (forall s. IOSim s a) -> a
runSimOrThrow ((forall s. IOSim s BlockFetchClientOutcome)
 -> BlockFetchClientOutcome)
-> (forall s. IOSim s BlockFetchClientOutcome)
-> BlockFetchClientOutcome
forall a b. (a -> b) -> a -> b
$ BlockFetchClientTestSetup -> IOSim s BlockFetchClientOutcome
forall (m :: * -> *).
(IOLike m, MonadTime m, MonadTimer m) =>
BlockFetchClientTestSetup -> m BlockFetchClientOutcome
runBlockFetchTest BlockFetchClientTestSetup
bfcts

    noException :: String -> Either e () -> Property
noException String
msg = \case
      Right () -> () -> Property
forall prop. Testable prop => prop -> Property
property ()
      Left e
ex  ->
        String -> Bool -> Property
forall prop. Testable prop => String -> prop -> Property
counterexample (String
msg String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
": exception: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> e -> String
forall e. Exception e => e -> String
displayException e
ex) Bool
False

    ppTrace :: (Tick, String) -> String
ppTrace (Tick Word64
tick, String
ev) = Word64 -> String
forall a. Show a => a -> String
show Word64
tick String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
": " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
ev

{-------------------------------------------------------------------------------
  Running a test involving the consensus BlockFetch interface
-------------------------------------------------------------------------------}

data BlockFetchClientOutcome = BlockFetchClientOutcome {
    BlockFetchClientOutcome -> Map PeerId (Either SomeException ())
bfcoBlockFetchResults :: Map PeerId (Either SomeException ())
  , BlockFetchClientOutcome -> Map PeerId Word
bfcoFetchedBlocks     :: Map PeerId Word
  , BlockFetchClientOutcome -> [(Tick, String)]
bfcoTrace             :: [(Tick, String)]
  }

runBlockFetchTest ::
     forall m.
     (IOLike m, MonadTime m, MonadTimer m)
  => BlockFetchClientTestSetup
  -> m BlockFetchClientOutcome
runBlockFetchTest :: forall (m :: * -> *).
(IOLike m, MonadTime m, MonadTimer m) =>
BlockFetchClientTestSetup -> m BlockFetchClientOutcome
runBlockFetchTest BlockFetchClientTestSetup{Map PeerId (Schedule ChainUpdate)
DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
peerUpdates :: BlockFetchClientTestSetup -> Map PeerId (Schedule ChainUpdate)
blockFetchMode :: BlockFetchClientTestSetup -> FetchMode
blockFetchCfg :: BlockFetchClientTestSetup -> BlockFetchConfiguration
blockFetchPipelining :: BlockFetchClientTestSetup -> DiffusionPipeliningSupport
peerUpdates :: Map PeerId (Schedule ChainUpdate)
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
..} = (ResourceRegistry m -> m BlockFetchClientOutcome)
-> m BlockFetchClientOutcome
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
withRegistry \ResourceRegistry m
registry -> do
    StrictTVar m (Map PeerId (Chain TestBlock))
varChains           <- Map PeerId (Chain TestBlock)
-> m (StrictTVar m (Map PeerId (Chain TestBlock)))
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
uncheckedNewTVarM Map PeerId (Chain TestBlock)
forall k a. Map k a
Map.empty
    StrictTVar m ControlMessage
varControlMessage   <- ControlMessage -> m (StrictTVar m ControlMessage)
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
uncheckedNewTVarM ControlMessage
Continue
    StrictTVar m (Map PeerId Word)
varFetchedBlocks    <- Map PeerId Word -> m (StrictTVar m (Map PeerId Word))
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
uncheckedNewTVarM (Word
0 Word -> Map PeerId (Schedule ChainUpdate) -> Map PeerId Word
forall a b. a -> Map PeerId b -> Map PeerId a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Map PeerId (Schedule ChainUpdate)
peerUpdates)

    FetchClientRegistry PeerId (Header TestBlock) TestBlock m
fetchClientRegistry <- m (FetchClientRegistry PeerId (Header TestBlock) TestBlock m)
forall (m :: * -> *) peer header block.
MonadSTM m =>
m (FetchClientRegistry peer header block m)
newFetchClientRegistry
    LogicalClock m
clock               <- ResourceRegistry m -> NumTicks -> m (LogicalClock m)
forall (m :: * -> *).
IOLike m =>
ResourceRegistry m -> NumTicks -> m (LogicalClock m)
LogicalClock.new ResourceRegistry m
registry (NumTicks -> m (LogicalClock m)) -> NumTicks -> m (LogicalClock m)
forall a b. (a -> b) -> a -> b
$
      HasCallStack => [Tick] -> NumTicks
[Tick] -> NumTicks
LogicalClock.sufficientTimeFor ([Tick] -> NumTicks) -> [Tick] -> NumTicks
forall a b. (a -> b) -> a -> b
$ Schedule ChainUpdate -> Tick
forall a. Schedule a -> Tick
lastTick (Schedule ChainUpdate -> Tick) -> [Schedule ChainUpdate] -> [Tick]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Map PeerId (Schedule ChainUpdate) -> [Schedule ChainUpdate]
forall k a. Map k a -> [a]
Map.elems Map PeerId (Schedule ChainUpdate)
peerUpdates
    (Tracer m String
tracer, m [(Tick, String)]
getTrace)  <-
      (Tracer m (Tick, String) -> Tracer m String)
-> (Tracer m (Tick, String), m [(Tick, String)])
-> (Tracer m String, m [(Tick, String)])
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first (LogicalClock m -> Tracer m (Tick, String) -> Tracer m String
forall (m :: * -> *) ev.
MonadSTM m =>
LogicalClock m -> Tracer m (Tick, ev) -> Tracer m ev
LogicalClock.tickTracer LogicalClock m
clock) ((Tracer m (Tick, String), m [(Tick, String)])
 -> (Tracer m String, m [(Tick, String)]))
-> m (Tracer m (Tick, String), m [(Tick, String)])
-> m (Tracer m String, m [(Tick, String)])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m (Tracer m (Tick, String), m [(Tick, String)])
forall (m :: * -> *) ev. MonadSTM m => m (Tracer m ev, m [ev])
recordingTracerTVar
    ChainDbView m TestBlock
chainDbView         <- ResourceRegistry m
-> Tracer m String -> m (ChainDbView m TestBlock)
mkChainDbView ResourceRegistry m
registry Tracer m String
tracer

    let getCandidates :: STM m (Map PeerId (AnchoredFragment TestBlock))
getCandidates = (Chain TestBlock -> AnchoredFragment TestBlock)
-> Map PeerId (Chain TestBlock)
-> Map PeerId (AnchoredFragment TestBlock)
forall a b k. (a -> b) -> Map k a -> Map k b
Map.map Chain TestBlock -> AnchoredFragment TestBlock
forall blk. HasHeader blk => Chain blk -> AnchoredFragment blk
chainToAnchoredFragment (Map PeerId (Chain TestBlock)
 -> Map PeerId (AnchoredFragment TestBlock))
-> STM m (Map PeerId (Chain TestBlock))
-> STM m (Map PeerId (AnchoredFragment TestBlock))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTVar m (Map PeerId (Chain TestBlock))
-> STM m (Map PeerId (Chain TestBlock))
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (Map PeerId (Chain TestBlock))
varChains

        blockFetchConsensusInterface :: BlockFetchConsensusInterface PeerId (Header TestBlock) TestBlock m
blockFetchConsensusInterface =
          STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
-> ChainDbView m TestBlock
-> BlockFetchConsensusInterface
     PeerId (Header TestBlock) TestBlock m
mkTestBlockFetchConsensusInterface
            ((AnchoredFragment TestBlock -> AnchoredFragment (Header TestBlock))
-> Map PeerId (AnchoredFragment TestBlock)
-> Map PeerId (AnchoredFragment (Header TestBlock))
forall a b k. (a -> b) -> Map k a -> Map k b
Map.map ((TestBlock -> Header TestBlock)
-> AnchoredFragment TestBlock
-> AnchoredFragment (Header TestBlock)
forall block2 block1.
(HasHeader block2, HeaderHash block1 ~ HeaderHash block2) =>
(block1 -> block2)
-> AnchoredFragment block1 -> AnchoredFragment block2
AF.mapAnchoredFragment TestBlock -> Header TestBlock
forall blk. GetHeader blk => blk -> Header blk
getHeader) (Map PeerId (AnchoredFragment TestBlock)
 -> Map PeerId (AnchoredFragment (Header TestBlock)))
-> STM m (Map PeerId (AnchoredFragment TestBlock))
-> STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM m (Map PeerId (AnchoredFragment TestBlock))
getCandidates)
            ChainDbView m TestBlock
chainDbView

    Thread m Void
_ <- ResourceRegistry m -> String -> m Void -> m (Thread m Void)
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, HasCallStack) =>
ResourceRegistry m -> String -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry m
registry String
"BlockFetchLogic" (m Void -> m (Thread m Void)) -> m Void -> m (Thread m Void)
forall a b. (a -> b) -> a -> b
$
      Tracer
  m
  [TraceLabelPeer PeerId (FetchDecision [Point (Header TestBlock)])]
-> Tracer
     m
     (TraceLabelPeer PeerId (TraceFetchClientState (Header TestBlock)))
-> BlockFetchConsensusInterface
     PeerId (Header TestBlock) TestBlock m
-> FetchClientRegistry PeerId (Header TestBlock) TestBlock m
-> BlockFetchConfiguration
-> m Void
forall addr header block (m :: * -> *).
(HasHeader header, HasHeader block,
 HeaderHash header ~ HeaderHash block, MonadDelay m, MonadSTM m,
 Ord addr, Hashable addr) =>
Tracer m [TraceLabelPeer addr (FetchDecision [Point header])]
-> Tracer m (TraceLabelPeer addr (TraceFetchClientState header))
-> BlockFetchConsensusInterface addr header block m
-> FetchClientRegistry addr header block m
-> BlockFetchConfiguration
-> m Void
blockFetchLogic
        Tracer
  m
  [TraceLabelPeer PeerId (FetchDecision [Point (Header TestBlock)])]
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer
        Tracer
  m
  (TraceLabelPeer PeerId (TraceFetchClientState (Header TestBlock)))
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer
        BlockFetchConsensusInterface PeerId (Header TestBlock) TestBlock m
blockFetchConsensusInterface
        FetchClientRegistry PeerId (Header TestBlock) TestBlock m
fetchClientRegistry
        BlockFetchConfiguration
blockFetchCfg

    let runBlockFetchClient :: PeerId -> m ()
runBlockFetchClient PeerId
peerId =
          FetchClientRegistry PeerId (Header TestBlock) TestBlock m
-> NodeToNodeVersion
-> PeerId
-> (FetchClientContext (Header TestBlock) TestBlock m -> m ())
-> m ()
forall (m :: * -> *) a peer header block version.
(MonadFork m, MonadMask m, MonadTimer m, Ord peer) =>
FetchClientRegistry peer header block m
-> version
-> peer
-> (FetchClientContext header block m -> m a)
-> m a
bracketFetchClient FetchClientRegistry PeerId (Header TestBlock) TestBlock m
fetchClientRegistry NodeToNodeVersion
ntnVersion PeerId
peerId \FetchClientContext (Header TestBlock) TestBlock m
clientCtx -> do
            let bfClient :: ClientPipelined
  (BlockFetch TestBlock (Point TestBlock)) 'BFIdle m ()
bfClient = NodeToNodeVersion
-> ControlMessageSTM m
-> FetchedMetricsTracer m
-> FetchClientContext (Header TestBlock) TestBlock m
-> ClientPipelined
     (BlockFetch TestBlock (Point TestBlock)) 'BFIdle m ()
forall header block versionNumber (m :: * -> *).
(MonadSTM m, MonadThrow m, MonadTime m, MonadMonotonicTime m,
 HasHeader header, HasHeader block,
 HeaderHash header ~ HeaderHash block) =>
versionNumber
-> ControlMessageSTM m
-> FetchedMetricsTracer m
-> FetchClientContext header block m
-> ClientPipelined (BlockFetch block (Point block)) 'BFIdle m ()
blockFetchClient
                    NodeToNodeVersion
ntnVersion
                    (StrictTVar m ControlMessage -> ControlMessageSTM m
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m ControlMessage
varControlMessage)
                    FetchedMetricsTracer m
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer
                    FetchClientContext (Header TestBlock) TestBlock m
clientCtx
                bfServer :: Server
  (BlockFetch TestBlock (Point TestBlock)) 'NonPipelined 'BFIdle m ()
bfServer =
                    BlockFetchServer TestBlock (Point TestBlock) m ()
-> Server
     (BlockFetch TestBlock (Point TestBlock)) 'NonPipelined 'BFIdle m ()
forall block point (m :: * -> *) a.
Functor m =>
BlockFetchServer block point m a
-> Server (BlockFetch block point) 'NonPipelined 'BFIdle m a
blockFetchServerPeer (BlockFetchServer TestBlock (Point TestBlock) m ()
 -> Server
      (BlockFetch TestBlock (Point TestBlock))
      'NonPipelined
      'BFIdle
      m
      ())
-> BlockFetchServer TestBlock (Point TestBlock) m ()
-> Server
     (BlockFetch TestBlock (Point TestBlock)) 'NonPipelined 'BFIdle m ()
forall a b. (a -> b) -> a -> b
$ m (AnchoredFragment TestBlock)
-> BlockFetchServer TestBlock (Point TestBlock) m ()
forall (m :: * -> *) blk.
(Monad m, HasHeader blk) =>
m (AnchoredFragment blk) -> BlockFetchServer blk (Point blk) m ()
mockBlockFetchServer m (AnchoredFragment TestBlock)
getCurrentChain
                  where
                    getCurrentChain :: m (AnchoredFragment TestBlock)
getCurrentChain = STM m (AnchoredFragment TestBlock)
-> m (AnchoredFragment TestBlock)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (AnchoredFragment TestBlock)
 -> m (AnchoredFragment TestBlock))
-> STM m (AnchoredFragment TestBlock)
-> m (AnchoredFragment TestBlock)
forall a b. (a -> b) -> a -> b
$ (Map PeerId (AnchoredFragment TestBlock)
-> PeerId -> AnchoredFragment TestBlock
forall k a. Ord k => Map k a -> k -> a
Map.! PeerId
peerId) (Map PeerId (AnchoredFragment TestBlock)
 -> AnchoredFragment TestBlock)
-> STM m (Map PeerId (AnchoredFragment TestBlock))
-> STM m (AnchoredFragment TestBlock)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM m (Map PeerId (AnchoredFragment TestBlock))
getCandidates

                blockFetchTracer :: Tracer m (PeerRole, Driver.TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
                blockFetchTracer :: Tracer
  m
  (PeerRole, TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
blockFetchTracer = ((PeerRole, TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
 -> m ())
-> Tracer
     m
     (PeerRole, TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
forall (m :: * -> *) a. (a -> m ()) -> Tracer m a
Tracer \case
                    (PeerRole
AsClient, TraceSendRecv (BlockFetch TestBlock (Point TestBlock))
ev) -> do
                      STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically case TraceSendRecv (BlockFetch TestBlock (Point TestBlock))
ev of
                        Driver.TraceRecvMsg (AnyMessage (MsgBlock block1
_)) ->
                           StrictTVar m (Map PeerId Word)
-> (Map PeerId Word -> Map PeerId Word) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m (Map PeerId Word)
varFetchedBlocks ((Map PeerId Word -> Map PeerId Word) -> STM m ())
-> (Map PeerId Word -> Map PeerId Word) -> STM m ()
forall a b. (a -> b) -> a -> b
$ (Word -> Word) -> PeerId -> Map PeerId Word -> Map PeerId Word
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust (Word -> Word -> Word
forall a. Num a => a -> a -> a
+ Word
1) PeerId
peerId
                        TraceSendRecv (BlockFetch TestBlock (Point TestBlock))
_ -> () -> STM m ()
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                      Tracer m String -> String -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m String
tracer (String -> m ()) -> String -> m ()
forall a b. (a -> b) -> a -> b
$
                        PeerId -> String
forall a. Show a => a -> String
show PeerId
peerId String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
": BlockFetchClient: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> TraceSendRecv (BlockFetch TestBlock (Point TestBlock)) -> String
forall a. Show a => a -> String
show TraceSendRecv (BlockFetch TestBlock (Point TestBlock))
ev
                    (PeerRole, TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            ((), ()) -> ()
forall a b. (a, b) -> a
fst (((), ()) -> ()) -> m ((), ()) -> m ()
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m (Channel m (AnyMessage (BlockFetch TestBlock (Point TestBlock))),
   Channel m (AnyMessage (BlockFetch TestBlock (Point TestBlock))))
-> Tracer
     m
     (PeerRole, TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
-> Codec
     (BlockFetch TestBlock (Point TestBlock))
     CodecFailure
     m
     (AnyMessage (BlockFetch TestBlock (Point TestBlock)))
-> ClientPipelined
     (BlockFetch TestBlock (Point TestBlock)) 'BFIdle m ()
-> Peer
     (BlockFetch TestBlock (Point TestBlock))
     (FlipAgency 'AsClient)
     'NonPipelined
     'BFIdle
     m
     ()
-> m ((), ())
forall (m :: * -> *) failure bytes ps (pr :: PeerRole) (st :: ps) a
       b.
(MonadAsync m, MonadCatch m, Exception failure) =>
m (Channel m bytes, Channel m bytes)
-> Tracer m (PeerRole, TraceSendRecv ps)
-> Codec ps failure m bytes
-> PeerPipelined ps pr st m a
-> Peer ps (FlipAgency pr) 'NonPipelined st m b
-> m (a, b)
Driver.runConnectedPeersPipelined
              m (Channel m (AnyMessage (BlockFetch TestBlock (Point TestBlock))),
   Channel m (AnyMessage (BlockFetch TestBlock (Point TestBlock))))
forall (m :: * -> *) a.
(MonadLabelledSTM m, MonadTraceSTM m, Show a) =>
m (Channel m a, Channel m a)
createConnectedChannels
              Tracer
  m
  (PeerRole, TraceSendRecv (BlockFetch TestBlock (Point TestBlock)))
blockFetchTracer
              Codec
  (BlockFetch TestBlock (Point TestBlock))
  CodecFailure
  m
  (AnyMessage (BlockFetch TestBlock (Point TestBlock)))
forall {k} {k1} (block :: k) (point :: k1) (m :: * -> *).
Monad m =>
Codec
  (BlockFetch block point)
  CodecFailure
  m
  (AnyMessage (BlockFetch block point))
codecBlockFetchId
              ClientPipelined
  (BlockFetch TestBlock (Point TestBlock)) 'BFIdle m ()
bfClient
              Peer
  (BlockFetch TestBlock (Point TestBlock))
  (FlipAgency 'AsClient)
  'NonPipelined
  'BFIdle
  m
  ()
Server
  (BlockFetch TestBlock (Point TestBlock)) 'NonPipelined 'BFIdle m ()
bfServer

        -- On every tick, we schedule updates to the shared chain fragment
        -- (mocking ChainSync).
        forkTicking :: PeerId -> m (Thread m Void)
forkTicking PeerId
peerId =
            ResourceRegistry m
-> String -> Watcher m Tick Tick -> m (Thread m Void)
forall (m :: * -> *) a fp.
(IOLike m, Eq fp, HasCallStack) =>
ResourceRegistry m -> String -> Watcher m a fp -> m (Thread m Void)
forkLinkedWatcher ResourceRegistry m
registry (String
"TickWatcher " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> PeerId -> String
forall a. Condense a => a -> String
condense PeerId
peerId) (Watcher m Tick Tick -> m (Thread m Void))
-> Watcher m Tick Tick -> m (Thread m Void)
forall a b. (a -> b) -> a -> b
$
              LogicalClock m -> (Tick -> m ()) -> Watcher m Tick Tick
forall (m :: * -> *).
LogicalClock m -> (Tick -> m ()) -> Watcher m Tick Tick
LogicalClock.tickWatcher LogicalClock m
clock \Tick
tick -> STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically do
                let updates :: [ChainUpdate TestBlock TestBlock]
updates = [ChainUpdate] -> [ChainUpdate TestBlock TestBlock]
toChainUpdates ([ChainUpdate] -> [ChainUpdate TestBlock TestBlock])
-> [ChainUpdate] -> [ChainUpdate TestBlock TestBlock]
forall a b. (a -> b) -> a -> b
$
                      [ChainUpdate] -> Tick -> Map Tick [ChainUpdate] -> [ChainUpdate]
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault [] Tick
tick (Map Tick [ChainUpdate] -> [ChainUpdate])
-> Map Tick [ChainUpdate] -> [ChainUpdate]
forall a b. (a -> b) -> a -> b
$
                        Schedule ChainUpdate -> Map Tick [ChainUpdate]
forall a. Schedule a -> Map Tick [a]
getSchedule (Schedule ChainUpdate -> Map Tick [ChainUpdate])
-> Schedule ChainUpdate -> Map Tick [ChainUpdate]
forall a b. (a -> b) -> a -> b
$ Map PeerId (Schedule ChainUpdate)
peerUpdates Map PeerId (Schedule ChainUpdate) -> PeerId -> Schedule ChainUpdate
forall k a. Ord k => Map k a -> k -> a
Map.! PeerId
peerId
                    updateChain :: Chain TestBlock -> Chain TestBlock
updateChain Chain TestBlock
chain =
                      case [ChainUpdate TestBlock TestBlock]
-> Chain TestBlock -> Maybe (Chain TestBlock)
forall block.
HasHeader block =>
[ChainUpdate block block] -> Chain block -> Maybe (Chain block)
Chain.applyChainUpdates [ChainUpdate TestBlock TestBlock]
updates Chain TestBlock
chain of
                        Just Chain TestBlock
chain' -> Chain TestBlock
chain'
                        Maybe (Chain TestBlock)
Nothing     -> String -> Chain TestBlock
forall a. HasCallStack => String -> a
error String
"Chain update failed"
                -- Block until our "ChainSync" thread registered itself to the
                -- FetchClientRegistry, see 'forkChainSync' below.
                Chain TestBlock
_ <- STM m (Maybe (Chain TestBlock)) -> STM m (Chain TestBlock)
forall (m :: * -> *) a. MonadSTM m => STM m (Maybe a) -> STM m a
blockUntilJust (STM m (Maybe (Chain TestBlock)) -> STM m (Chain TestBlock))
-> STM m (Maybe (Chain TestBlock)) -> STM m (Chain TestBlock)
forall a b. (a -> b) -> a -> b
$ PeerId -> Map PeerId (Chain TestBlock) -> Maybe (Chain TestBlock)
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
peerId (Map PeerId (Chain TestBlock) -> Maybe (Chain TestBlock))
-> STM m (Map PeerId (Chain TestBlock))
-> STM m (Maybe (Chain TestBlock))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTVar m (Map PeerId (Chain TestBlock))
-> STM m (Map PeerId (Chain TestBlock))
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (Map PeerId (Chain TestBlock))
varChains
                StrictTVar m (Map PeerId (Chain TestBlock))
-> (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m (Map PeerId (Chain TestBlock))
varChains ((Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
 -> STM m ())
-> (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> STM m ()
forall a b. (a -> b) -> a -> b
$ (Chain TestBlock -> Chain TestBlock)
-> PeerId
-> Map PeerId (Chain TestBlock)
-> Map PeerId (Chain TestBlock)
forall k a. Ord k => (a -> a) -> k -> Map k a -> Map k a
Map.adjust Chain TestBlock -> Chain TestBlock
updateChain PeerId
peerId

        forkChainSync :: PeerId -> m (Thread m Void)
forkChainSync PeerId
peerId =
          ResourceRegistry m -> String -> m Void -> m (Thread m Void)
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, HasCallStack) =>
ResourceRegistry m -> String -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry m
registry (String
"BracketSync" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> PeerId -> String
forall a. Condense a => a -> String
condense PeerId
peerId) (m Void -> m (Thread m Void)) -> m Void -> m (Thread m Void)
forall a b. (a -> b) -> a -> b
$
            FetchClientRegistry PeerId (Header TestBlock) TestBlock m
-> PeerId -> m Void -> m Void
forall (m :: * -> *) a peer header block.
(MonadSTM m, MonadFork m, MonadCatch m, Ord peer) =>
FetchClientRegistry peer header block m -> peer -> m a -> m a
bracketSyncWithFetchClient FetchClientRegistry PeerId (Header TestBlock) TestBlock m
fetchClientRegistry PeerId
peerId (m Void -> m Void) -> m Void -> m Void
forall a b. (a -> b) -> a -> b
$ do
              let modifyChains :: (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> m ()
modifyChains = STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ())
-> ((Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
    -> STM m ())
-> (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StrictTVar m (Map PeerId (Chain TestBlock))
-> (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m (Map PeerId (Chain TestBlock))
varChains
              m () -> m () -> m Void -> m Void
forall a b c. m a -> m b -> m c -> m c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> m b -> m c -> m c
bracket_
                ((Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> m ()
modifyChains ((Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
 -> m ())
-> (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> m ()
forall a b. (a -> b) -> a -> b
$ PeerId
-> Chain TestBlock
-> Map PeerId (Chain TestBlock)
-> Map PeerId (Chain TestBlock)
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
peerId Chain TestBlock
forall block. Chain block
Chain.Genesis)
                ((Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> m ()
modifyChains ((Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
 -> m ())
-> (Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock))
-> m ()
forall a b. (a -> b) -> a -> b
$ PeerId
-> Map PeerId (Chain TestBlock) -> Map PeerId (Chain TestBlock)
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete PeerId
peerId)
                (PeerId -> m (Thread m Void)
forkTicking PeerId
peerId m (Thread m Void) -> (Thread m Void -> 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
>>= Thread m Void -> m Void
forall (m :: * -> *) a. MonadAsync m => Thread m a -> m a
waitThread)

        -- The BlockFetch logic requires initializing the KeepAlive
        -- miniprotocol, even if it does not do anything.
        forkKeepAlive :: PeerId -> m (Thread m Any)
forkKeepAlive PeerId
peerId =
          ResourceRegistry m -> String -> m Any -> m (Thread m Any)
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, HasCallStack) =>
ResourceRegistry m -> String -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry m
registry String
"KeepAlive" (m Any -> m (Thread m Any)) -> m Any -> m (Thread m Any)
forall a b. (a -> b) -> a -> b
$
            FetchClientRegistry PeerId (Header TestBlock) TestBlock m
-> PeerId -> (StrictTVar m (Map PeerId PeerGSV) -> m Any) -> m Any
forall (m :: * -> *) a peer header block.
(MonadSTM m, MonadFork m, MonadMask m, Ord peer) =>
FetchClientRegistry peer header block m
-> peer -> (StrictTVar m (Map peer PeerGSV) -> m a) -> m a
bracketKeepAliveClient FetchClientRegistry PeerId (Header TestBlock) TestBlock m
fetchClientRegistry PeerId
peerId \StrictTVar m (Map PeerId PeerGSV)
_ ->
              m Any
forall (m :: * -> *) a. MonadSTM m => m a
infiniteDelay

    Map PeerId (Thread m (Either SomeException ()))
blockFetchThreads <- [(PeerId, Thread m (Either SomeException ()))]
-> Map PeerId (Thread m (Either SomeException ()))
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(PeerId, Thread m (Either SomeException ()))]
 -> Map PeerId (Thread m (Either SomeException ())))
-> m [(PeerId, Thread m (Either SomeException ()))]
-> m (Map PeerId (Thread m (Either SomeException ())))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [PeerId]
-> (PeerId -> m (PeerId, Thread m (Either SomeException ())))
-> m [(PeerId, Thread m (Either SomeException ()))]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
t a -> (a -> f b) -> f (t b)
for [PeerId]
peerIds \PeerId
peerId -> do
      Thread m Void
_ <- PeerId -> m (Thread m Void)
forkChainSync PeerId
peerId
      Thread m Any
_ <- PeerId -> m (Thread m Any)
forkKeepAlive PeerId
peerId
      (Thread m (Either SomeException ())
 -> (PeerId, Thread m (Either SomeException ())))
-> m (Thread m (Either SomeException ()))
-> m (PeerId, Thread m (Either SomeException ()))
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (PeerId
peerId,) (m (Thread m (Either SomeException ()))
 -> m (PeerId, Thread m (Either SomeException ())))
-> m (Thread m (Either SomeException ()))
-> m (PeerId, Thread m (Either SomeException ()))
forall a b. (a -> b) -> a -> b
$
        ResourceRegistry m
-> String
-> m (Either SomeException ())
-> m (Thread m (Either SomeException ()))
forall (m :: * -> *) a.
(MonadMask m, MonadAsync m, HasCallStack) =>
ResourceRegistry m -> String -> m a -> m (Thread m a)
forkThread ResourceRegistry m
registry (String
"BlockFetch " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> PeerId -> String
forall a. Condense a => a -> String
condense PeerId
peerId) (m (Either SomeException ())
 -> m (Thread m (Either SomeException ())))
-> m (Either SomeException ())
-> m (Thread m (Either SomeException ()))
forall a b. (a -> b) -> a -> b
$
          m () -> m (Either SomeException ())
forall e a. Exception e => m a -> m (Either e a)
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> m (Either e a)
try (m () -> m (Either SomeException ()))
-> m () -> m (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ PeerId -> m ()
runBlockFetchClient PeerId
peerId

    LogicalClock m -> m ()
forall (m :: * -> *). LogicalClock m -> m ()
LogicalClock.waitUntilDone LogicalClock m
clock
    STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ StrictTVar m ControlMessage -> ControlMessage -> STM m ()
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar m ControlMessage
varControlMessage ControlMessage
Terminate

    Map PeerId (Either SomeException ())
bfcoBlockFetchResults <- (Thread m (Either SomeException ()) -> m (Either SomeException ()))
-> Map PeerId (Thread m (Either SomeException ()))
-> m (Map PeerId (Either SomeException ()))
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> Map PeerId a -> f (Map PeerId b)
traverse Thread m (Either SomeException ()) -> m (Either SomeException ())
forall (m :: * -> *) a. MonadAsync m => Thread m a -> m a
waitThread Map PeerId (Thread m (Either SomeException ()))
blockFetchThreads
    Map PeerId Word
bfcoFetchedBlocks     <- StrictTVar m (Map PeerId Word) -> m (Map PeerId Word)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> m a
readTVarIO StrictTVar m (Map PeerId Word)
varFetchedBlocks
    [(Tick, String)]
bfcoTrace             <- m [(Tick, String)]
getTrace
    BlockFetchClientOutcome -> m BlockFetchClientOutcome
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure BlockFetchClientOutcome {[(Tick, String)]
Map PeerId Word
Map PeerId (Either SomeException ())
bfcoBlockFetchResults :: Map PeerId (Either SomeException ())
bfcoFetchedBlocks :: Map PeerId Word
bfcoTrace :: [(Tick, String)]
bfcoBlockFetchResults :: Map PeerId (Either SomeException ())
bfcoFetchedBlocks :: Map PeerId Word
bfcoTrace :: [(Tick, String)]
..}
  where
    peerIds :: [PeerId]
peerIds = Map PeerId (Schedule ChainUpdate) -> [PeerId]
forall k a. Map k a -> [k]
Map.keys Map PeerId (Schedule ChainUpdate)
peerUpdates

    numCoreNodes :: NumCoreNodes
numCoreNodes = Word64 -> NumCoreNodes
NumCoreNodes (Word64 -> NumCoreNodes) -> Word64 -> NumCoreNodes
forall a b. (a -> b) -> a -> b
$ Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Word64) -> Int -> Word64
forall a b. (a -> b) -> a -> b
$ Map PeerId (Schedule ChainUpdate) -> Int
forall k a. Map k a -> Int
Map.size Map PeerId (Schedule ChainUpdate)
peerUpdates Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1

    mkChainDbView ::
         ResourceRegistry m
      -> Tracer m String
      -> m (BlockFetchClientInterface.ChainDbView m TestBlock)
    mkChainDbView :: ResourceRegistry m
-> Tracer m String -> m (ChainDbView m TestBlock)
mkChainDbView ResourceRegistry m
registry Tracer m String
tracer = do
        ChainDbArgs Identity m TestBlock
chainDbArgs <- do
          NodeDBs (StrictTMVar m MockFS)
nodeDBs <- m (NodeDBs (StrictTMVar m MockFS))
forall (m :: * -> *).
MonadSTM m =>
m (NodeDBs (StrictTMVar m MockFS))
emptyNodeDBs
          let args :: ChainDbArgs Identity m TestBlock
args = MinimalChainDbArgs m TestBlock -> ChainDbArgs Identity m TestBlock
forall (m :: * -> *) blk.
(MonadThrow m, MonadSTM m, ConsensusProtocol (BlockProtocol blk),
 PrimMonad m) =>
MinimalChainDbArgs m blk -> Complete ChainDbArgs m blk
fromMinimalChainDbArgs (MinimalChainDbArgs m TestBlock
 -> ChainDbArgs Identity m TestBlock)
-> MinimalChainDbArgs m TestBlock
-> ChainDbArgs Identity m TestBlock
forall a b. (a -> b) -> a -> b
$ MinimalChainDbArgs {
                  mcdbTopLevelConfig :: TopLevelConfig TestBlock
mcdbTopLevelConfig = TopLevelConfig TestBlock
topLevelConfig
                , mcdbChunkInfo :: ChunkInfo
mcdbChunkInfo = TopLevelConfig TestBlock -> ChunkInfo
mkTestChunkInfo TopLevelConfig TestBlock
topLevelConfig
                , mcdbInitLedger :: ExtLedgerState TestBlock
mcdbInitLedger = ExtLedgerState TestBlock
testInitExtLedger
                , mcdbRegistry :: ResourceRegistry m
mcdbRegistry = ResourceRegistry m
registry
                , mcdbNodeDBs :: NodeDBs (StrictTMVar m MockFS)
mcdbNodeDBs = NodeDBs (StrictTMVar m MockFS)
nodeDBs
                }
          ChainDbArgs Identity m TestBlock
-> m (ChainDbArgs Identity m TestBlock)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ChainDbArgs Identity m TestBlock
 -> m (ChainDbArgs Identity m TestBlock))
-> ChainDbArgs Identity m TestBlock
-> m (ChainDbArgs Identity m TestBlock)
forall a b. (a -> b) -> a -> b
$ Tracer m (TraceEvent TestBlock)
-> ChainDbArgs Identity m TestBlock
-> ChainDbArgs Identity m TestBlock
forall (m :: * -> *) blk (f :: * -> *).
Tracer m (TraceEvent blk)
-> ChainDbArgs f m blk -> ChainDbArgs f m blk
ChainDB.updateTracer Tracer m (TraceEvent TestBlock)
cdbTracer ChainDbArgs Identity m TestBlock
args
        (ResourceKey m
_, (ChainDB m TestBlock
chainDB, ChainDBImpl.Internal{m Void
intAddBlockRunner :: m Void
intAddBlockRunner :: forall (m :: * -> *) blk. Internal m blk -> m Void
intAddBlockRunner})) <-
          ResourceRegistry m
-> (ResourceId -> m (ChainDB m TestBlock, Internal m TestBlock))
-> ((ChainDB m TestBlock, Internal m TestBlock) -> m ())
-> m (ResourceKey m, (ChainDB m TestBlock, Internal m TestBlock))
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, HasCallStack) =>
ResourceRegistry m
-> (ResourceId -> m a) -> (a -> m ()) -> m (ResourceKey m, a)
allocate
            ResourceRegistry m
registry
            (\ResourceId
_ -> ChainDbArgs Identity m TestBlock
-> Bool -> m (ChainDB m TestBlock, Internal m TestBlock)
forall (m :: * -> *) blk.
(IOLike m, LedgerSupportsProtocol blk,
 BlockSupportsDiffusionPipelining blk, InspectLedger blk,
 HasHardForkHistory blk, ConvertRawHash blk,
 SerialiseDiskConstraints blk) =>
ChainDbArgs Identity m blk
-> Bool -> m (ChainDB m blk, Internal m blk)
ChainDBImpl.openDBInternal ChainDbArgs Identity m TestBlock
chainDbArgs Bool
False)
            (ChainDB m TestBlock -> m ()
forall (m :: * -> *) blk. ChainDB m blk -> m ()
ChainDB.closeDB (ChainDB m TestBlock -> m ())
-> ((ChainDB m TestBlock, Internal m TestBlock)
    -> ChainDB m TestBlock)
-> (ChainDB m TestBlock, Internal m TestBlock)
-> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (ChainDB m TestBlock, Internal m TestBlock) -> ChainDB m TestBlock
forall a b. (a, b) -> a
fst)
        Thread m Void
_ <- ResourceRegistry m -> String -> m Void -> m (Thread m Void)
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, HasCallStack) =>
ResourceRegistry m -> String -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry m
registry String
"AddBlockRunner" m Void
intAddBlockRunner

        let -- Always return the empty chain such that the BlockFetch logic
            -- downloads all chains.
            getCurrentChain :: STM m (AnchoredFragment (Header TestBlock))
getCurrentChain           = AnchoredFragment (Header TestBlock)
-> STM m (AnchoredFragment (Header TestBlock))
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (AnchoredFragment (Header TestBlock)
 -> STM m (AnchoredFragment (Header TestBlock)))
-> AnchoredFragment (Header TestBlock)
-> STM m (AnchoredFragment (Header TestBlock))
forall a b. (a -> b) -> a -> b
$ Anchor (Header TestBlock) -> AnchoredFragment (Header TestBlock)
forall v a b. Anchorable v a b => a -> AnchoredSeq v a b
AF.Empty Anchor (Header TestBlock)
forall block. Anchor block
AF.AnchorGenesis
            getIsFetched :: STM m (Point TestBlock -> Bool)
getIsFetched              = ChainDB m TestBlock -> STM m (Point TestBlock -> Bool)
forall (m :: * -> *) blk.
ChainDB m blk -> STM m (Point blk -> Bool)
ChainDB.getIsFetched ChainDB m TestBlock
chainDB
            getMaxSlotNo :: STM m MaxSlotNo
getMaxSlotNo              = ChainDB m TestBlock -> STM m MaxSlotNo
forall (m :: * -> *) blk. ChainDB m blk -> STM m MaxSlotNo
ChainDB.getMaxSlotNo ChainDB m TestBlock
chainDB
            addBlockWaitWrittenToDisk :: InvalidBlockPunishment m -> TestBlock -> m Bool
addBlockWaitWrittenToDisk = ChainDB m TestBlock
-> InvalidBlockPunishment m -> TestBlock -> m Bool
forall (m :: * -> *) blk.
IOLike m =>
ChainDB m blk -> InvalidBlockPunishment m -> blk -> m Bool
ChainDB.addBlockWaitWrittenToDisk ChainDB m TestBlock
chainDB
        ChainDbView m TestBlock -> m (ChainDbView m TestBlock)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure BlockFetchClientInterface.ChainDbView {STM m (AnchoredFragment (Header TestBlock))
STM m MaxSlotNo
STM m (Point TestBlock -> Bool)
InvalidBlockPunishment m -> TestBlock -> m Bool
getCurrentChain :: STM m (AnchoredFragment (Header TestBlock))
getIsFetched :: STM m (Point TestBlock -> Bool)
getMaxSlotNo :: STM m MaxSlotNo
addBlockWaitWrittenToDisk :: InvalidBlockPunishment m -> TestBlock -> m Bool
getCurrentChain :: STM m (AnchoredFragment (Header TestBlock))
getIsFetched :: STM m (Point TestBlock -> Bool)
getMaxSlotNo :: STM m MaxSlotNo
addBlockWaitWrittenToDisk :: InvalidBlockPunishment m -> TestBlock -> m Bool
..}
      where
        -- Needs to be larger than any chain length in this test, to ensure that
        -- switching to any chain is never too deep.
        securityParam :: SecurityParam
securityParam  = Word64 -> SecurityParam
SecurityParam Word64
1000
        topLevelConfig :: TopLevelConfig TestBlock
topLevelConfig = SecurityParam -> TopLevelConfig TestBlock
singleNodeTestConfigWithK SecurityParam
securityParam

        cdbTracer :: Tracer m (TraceEvent TestBlock)
cdbTracer = (TraceEvent TestBlock -> m ()) -> Tracer m (TraceEvent TestBlock)
forall (m :: * -> *) a. (a -> m ()) -> Tracer m a
Tracer \case
            ChainDBImpl.TraceAddBlockEvent TraceAddBlockEvent TestBlock
ev ->
              Tracer m String -> String -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m String
tracer (String -> m ()) -> String -> m ()
forall a b. (a -> b) -> a -> b
$ String
"ChainDB: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> TraceAddBlockEvent TestBlock -> String
forall a. Show a => a -> String
show TraceAddBlockEvent TestBlock
ev
            TraceEvent TestBlock
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()


    mkTestBlockFetchConsensusInterface ::
         STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
      -> BlockFetchClientInterface.ChainDbView m TestBlock
      -> BlockFetchConsensusInterface PeerId (Header TestBlock) TestBlock m
    mkTestBlockFetchConsensusInterface :: STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
-> ChainDbView m TestBlock
-> BlockFetchConsensusInterface
     PeerId (Header TestBlock) TestBlock m
mkTestBlockFetchConsensusInterface STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
getCandidates ChainDbView m TestBlock
chainDbView =
        BlockConfig TestBlock
-> ChainDbView m TestBlock
-> STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
-> (Header TestBlock -> SizeInBytes)
-> SlotForgeTimeOracle m TestBlock
-> STM m FetchMode
-> DiffusionPipeliningSupport
-> BlockFetchConsensusInterface
     PeerId (Header TestBlock) TestBlock m
forall (m :: * -> *) peer blk.
(IOLike m, BlockSupportsDiffusionPipelining blk,
 BlockSupportsProtocol blk) =>
BlockConfig blk
-> ChainDbView m blk
-> STM m (Map peer (AnchoredFragment (Header blk)))
-> (Header blk -> SizeInBytes)
-> SlotForgeTimeOracle m blk
-> STM m FetchMode
-> DiffusionPipeliningSupport
-> BlockFetchConsensusInterface peer (Header blk) blk m
BlockFetchClientInterface.mkBlockFetchConsensusInterface
          (NumCoreNodes -> BlockConfig TestBlock
forall ptype. NumCoreNodes -> BlockConfig (TestBlockWith ptype)
TestBlockConfig NumCoreNodes
numCoreNodes)
          ChainDbView m TestBlock
chainDbView
          STM m (Map PeerId (AnchoredFragment (Header TestBlock)))
getCandidates
          (\Header TestBlock
_hdr -> SizeInBytes
1000) -- header size, only used for peer prioritization
          SlotForgeTimeOracle m TestBlock
forall blk. SlotForgeTimeOracle m blk
slotForgeTime
          (FetchMode -> STM m FetchMode
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure FetchMode
blockFetchMode)
          DiffusionPipeliningSupport
blockFetchPipelining
      where
        -- Bogus implementation; this is fine as this is only used for
        -- enriching tracing information ATM.
        slotForgeTime :: BlockFetchClientInterface.SlotForgeTimeOracle m blk
        slotForgeTime :: forall blk. SlotForgeTimeOracle m blk
slotForgeTime RealPoint blk
_ = UTCTime -> STM m UTCTime
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure UTCTime
dawnOfTime

mockBlockFetchServer ::
     forall m blk.
     (Monad m, HasHeader blk)
  => m (AnchoredFragment blk)
  -> BlockFetchServer blk (Point blk) m ()
mockBlockFetchServer :: forall (m :: * -> *) blk.
(Monad m, HasHeader blk) =>
m (AnchoredFragment blk) -> BlockFetchServer blk (Point blk) m ()
mockBlockFetchServer m (AnchoredFragment blk)
getCurrentChain = BlockFetchServer blk (Point blk) m ()
idle
  where
    idle :: BlockFetchServer blk (Point blk) m ()
    idle :: BlockFetchServer blk (Point blk) m ()
idle = ((ChainRange (Point blk)
  -> m (BlockFetchBlockSender blk (Point blk) m ()))
 -> () -> BlockFetchServer blk (Point blk) m ())
-> ()
-> (ChainRange (Point blk)
    -> m (BlockFetchBlockSender blk (Point blk) m ()))
-> BlockFetchServer blk (Point blk) m ()
forall a b c. (a -> b -> c) -> b -> a -> c
flip (ChainRange (Point blk)
 -> m (BlockFetchBlockSender blk (Point blk) m ()))
-> () -> BlockFetchServer blk (Point blk) m ()
forall point (m :: * -> *) block a.
(ChainRange point -> m (BlockFetchBlockSender block point m a))
-> a -> BlockFetchServer block point m a
BlockFetchServer () \(ChainRange Point blk
from Point blk
to) -> do
        AnchoredFragment blk
curChain <- m (AnchoredFragment blk)
getCurrentChain
        BlockFetchBlockSender blk (Point blk) m ()
-> m (BlockFetchBlockSender blk (Point blk) m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure case AnchoredFragment blk
-> Point blk -> Point blk -> Maybe (AnchoredFragment blk)
forall block.
HasHeader block =>
AnchoredFragment block
-> Point block -> Point block -> Maybe (AnchoredFragment block)
AF.sliceRange AnchoredFragment blk
curChain Point blk
from Point blk
to of
          Maybe (AnchoredFragment blk)
Nothing    -> m (BlockFetchServer blk (Point blk) m ())
-> BlockFetchBlockSender blk (Point blk) m ()
forall (m :: * -> *) block point a.
m (BlockFetchServer block point m a)
-> BlockFetchBlockSender block point m a
SendMsgNoBlocks (BlockFetchServer blk (Point blk) m ()
-> m (BlockFetchServer blk (Point blk) m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure BlockFetchServer blk (Point blk) m ()
idle)
          Just AnchoredFragment blk
slice -> m (BlockFetchSendBlocks blk (Point blk) m ())
-> BlockFetchBlockSender blk (Point blk) m ()
forall (m :: * -> *) block point a.
m (BlockFetchSendBlocks block point m a)
-> BlockFetchBlockSender block point m a
SendMsgStartBatch (m (BlockFetchSendBlocks blk (Point blk) m ())
 -> BlockFetchBlockSender blk (Point blk) m ())
-> m (BlockFetchSendBlocks blk (Point blk) m ())
-> BlockFetchBlockSender blk (Point blk) m ()
forall a b. (a -> b) -> a -> b
$ [blk] -> m (BlockFetchSendBlocks blk (Point blk) m ())
sendBlocks (AnchoredFragment blk -> [blk]
forall v a b. AnchoredSeq v a b -> [b]
AF.toOldestFirst AnchoredFragment blk
slice)

    sendBlocks :: [blk] -> m (BlockFetchSendBlocks blk (Point blk) m ())
    sendBlocks :: [blk] -> m (BlockFetchSendBlocks blk (Point blk) m ())
sendBlocks = BlockFetchSendBlocks blk (Point blk) m ()
-> m (BlockFetchSendBlocks blk (Point blk) m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BlockFetchSendBlocks blk (Point blk) m ()
 -> m (BlockFetchSendBlocks blk (Point blk) m ()))
-> ([blk] -> BlockFetchSendBlocks blk (Point blk) m ())
-> [blk]
-> m (BlockFetchSendBlocks blk (Point blk) m ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. \case
      []         -> m (BlockFetchServer blk (Point blk) m ())
-> BlockFetchSendBlocks blk (Point blk) m ()
forall (m :: * -> *) block point a.
m (BlockFetchServer block point m a)
-> BlockFetchSendBlocks block point m a
SendMsgBatchDone (BlockFetchServer blk (Point blk) m ()
-> m (BlockFetchServer blk (Point blk) m ())
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure BlockFetchServer blk (Point blk) m ()
idle)
      blk
blk : [blk]
blks -> blk
-> m (BlockFetchSendBlocks blk (Point blk) m ())
-> BlockFetchSendBlocks blk (Point blk) m ()
forall block (m :: * -> *) point a.
block
-> m (BlockFetchSendBlocks block point m a)
-> BlockFetchSendBlocks block point m a
SendMsgBlock blk
blk ([blk] -> m (BlockFetchSendBlocks blk (Point blk) m ())
sendBlocks [blk]
blks)

ntnVersion :: NodeToNodeVersion
ntnVersion :: NodeToNodeVersion
ntnVersion = NodeToNodeVersion
forall a. Bounded a => a
maxBound

{-------------------------------------------------------------------------------
  BlockFetchClientTestSetup
-------------------------------------------------------------------------------}

data BlockFetchClientTestSetup = BlockFetchClientTestSetup {
    -- | A 'Schedule' of 'ChainUpdate's for every peer. This emulates
    -- the candidate fragments provided by the ChainSync client.
    BlockFetchClientTestSetup -> Map PeerId (Schedule ChainUpdate)
peerUpdates          :: Map PeerId (Schedule ChainUpdate)
    -- | BlockFetch 'FetchMode'
  , BlockFetchClientTestSetup -> FetchMode
blockFetchMode       :: FetchMode
  , BlockFetchClientTestSetup -> BlockFetchConfiguration
blockFetchCfg        :: BlockFetchConfiguration
  , BlockFetchClientTestSetup -> DiffusionPipeliningSupport
blockFetchPipelining :: DiffusionPipeliningSupport
  }
  deriving stock (Int -> BlockFetchClientTestSetup -> String -> String
[BlockFetchClientTestSetup] -> String -> String
BlockFetchClientTestSetup -> String
(Int -> BlockFetchClientTestSetup -> String -> String)
-> (BlockFetchClientTestSetup -> String)
-> ([BlockFetchClientTestSetup] -> String -> String)
-> Show BlockFetchClientTestSetup
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> BlockFetchClientTestSetup -> String -> String
showsPrec :: Int -> BlockFetchClientTestSetup -> String -> String
$cshow :: BlockFetchClientTestSetup -> String
show :: BlockFetchClientTestSetup -> String
$cshowList :: [BlockFetchClientTestSetup] -> String -> String
showList :: [BlockFetchClientTestSetup] -> String -> String
Show)

instance Condense BlockFetchClientTestSetup where
  condense :: BlockFetchClientTestSetup -> String
condense BlockFetchClientTestSetup{Map PeerId (Schedule ChainUpdate)
DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
peerUpdates :: BlockFetchClientTestSetup -> Map PeerId (Schedule ChainUpdate)
blockFetchMode :: BlockFetchClientTestSetup -> FetchMode
blockFetchCfg :: BlockFetchClientTestSetup -> BlockFetchConfiguration
blockFetchPipelining :: BlockFetchClientTestSetup -> DiffusionPipeliningSupport
peerUpdates :: Map PeerId (Schedule ChainUpdate)
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
..} = [String] -> String
unlines
      [ String
"Number of peers: "
          String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int -> String
forall a. Show a => a -> String
show (Map PeerId (Schedule ChainUpdate) -> Int
forall k a. Map k a -> Int
Map.size Map PeerId (Schedule ChainUpdate)
peerUpdates)
      , String
"Chain updates:\n"
          String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Map PeerId (Schedule ChainUpdate) -> String
forall {a} {a}. (Condense a, Show a) => Map a a -> String
ppPerPeer Map PeerId (Schedule ChainUpdate)
peerUpdates
      , String
"BlockFetch mode: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> FetchMode -> String
forall a. Show a => a -> String
show FetchMode
blockFetchMode
      , String
"BlockFetch pipelining " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> DiffusionPipeliningSupport -> String
forall a. Show a => a -> String
show DiffusionPipeliningSupport
blockFetchPipelining
      , String
"BlockFetch cfg: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> BlockFetchConfiguration -> String
forall a. Show a => a -> String
show BlockFetchConfiguration
blockFetchCfg
      ]
    where
      ppPerPeer :: Map a a -> String
ppPerPeer Map a a
peerMap = [String] -> String
unlines
        [ String
"  " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> a -> String
forall a. Show a => a -> String
show a
peerId String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
": " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
valLine
        | (a
peerId, a
val) <- Map a a -> [(a, a)]
forall k a. Map k a -> [(k, a)]
Map.toAscList Map a a
peerMap
        , String
valLine       <- String -> [String]
lines (String -> [String]) -> String -> [String]
forall a b. (a -> b) -> a -> b
$ a -> String
forall a. Condense a => a -> String
condense a
val
        ]

instance Arbitrary BlockFetchClientTestSetup where
  arbitrary :: Gen BlockFetchClientTestSetup
arbitrary = do
      Int
numPeers <- (Int, Int) -> Gen Int
chooseInt (Int
1, Int
3)
      let peerIds :: [PeerId]
peerIds = Int -> PeerId
PeerId (Int -> PeerId) -> [Int] -> [PeerId]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Int
1 .. Int
numPeers]
      DiffusionPipeliningSupport
blockFetchPipelining <-
        [DiffusionPipeliningSupport] -> Gen DiffusionPipeliningSupport
forall a. HasCallStack => [a] -> Gen a
elements [DiffusionPipeliningSupport
DiffusionPipeliningOn, DiffusionPipeliningSupport
DiffusionPipeliningOff]
      Map PeerId (Schedule ChainUpdate)
peerUpdates <-
            [(PeerId, Schedule ChainUpdate)]
-> Map PeerId (Schedule ChainUpdate)
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(PeerId, Schedule ChainUpdate)]
 -> Map PeerId (Schedule ChainUpdate))
-> ([Schedule ChainUpdate] -> [(PeerId, Schedule ChainUpdate)])
-> [Schedule ChainUpdate]
-> Map PeerId (Schedule ChainUpdate)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [PeerId]
-> [Schedule ChainUpdate] -> [(PeerId, Schedule ChainUpdate)]
forall a b. [a] -> [b] -> [(a, b)]
zip [PeerId]
peerIds
        ([Schedule ChainUpdate] -> Map PeerId (Schedule ChainUpdate))
-> Gen [Schedule ChainUpdate]
-> Gen (Map PeerId (Schedule ChainUpdate))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Gen (Schedule ChainUpdate) -> Gen [Schedule ChainUpdate]
forall (m :: * -> *) a. Applicative m => Int -> m a -> m [a]
replicateM Int
numPeers (DiffusionPipeliningSupport -> Gen (Schedule ChainUpdate)
genUpdateSchedule DiffusionPipeliningSupport
blockFetchPipelining)
      FetchMode
blockFetchMode <- [FetchMode] -> Gen FetchMode
forall a. HasCallStack => [a] -> Gen a
elements [FetchMode
FetchModeBulkSync, FetchMode
FetchModeDeadline]
      BlockFetchConfiguration
blockFetchCfg  <- do
        let -- ensure that we can download blocks from all peers
            bfcMaxConcurrencyBulkSync :: Word
bfcMaxConcurrencyBulkSync = Int -> Word
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
numPeers
            bfcMaxConcurrencyDeadline :: Word
bfcMaxConcurrencyDeadline = Int -> Word
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
numPeers
            -- This is used to introduce a minimal delay between BlockFetch
            -- logic iterations in case the monitored state vars change too
            -- fast, which we don't have to worry about in this test.
            bfcDecisionLoopInterval :: DiffTime
bfcDecisionLoopInterval   = DiffTime
0
        Word
bfcMaxRequestsInflight <- (Word, Word) -> Gen Word
forall a. Enum a => (a, a) -> Gen a
chooseEnum (Word
2, Word
10)
        Int
bfcSalt                <- Gen Int
forall a. Arbitrary a => Gen a
arbitrary
        BlockFetchConfiguration -> Gen BlockFetchConfiguration
forall a. a -> Gen a
forall (f :: * -> *) a. Applicative f => a -> f a
pure BlockFetchConfiguration {Int
Word
DiffTime
bfcMaxConcurrencyBulkSync :: Word
bfcMaxConcurrencyDeadline :: Word
bfcDecisionLoopInterval :: DiffTime
bfcMaxRequestsInflight :: Word
bfcSalt :: Int
bfcMaxConcurrencyBulkSync :: Word
bfcMaxConcurrencyDeadline :: Word
bfcMaxRequestsInflight :: Word
bfcDecisionLoopInterval :: DiffTime
bfcSalt :: Int
..}
      BlockFetchClientTestSetup -> Gen BlockFetchClientTestSetup
forall a. a -> Gen a
forall (f :: * -> *) a. Applicative f => a -> f a
pure BlockFetchClientTestSetup {Map PeerId (Schedule ChainUpdate)
DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
peerUpdates :: Map PeerId (Schedule ChainUpdate)
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
blockFetchPipelining :: DiffusionPipeliningSupport
peerUpdates :: Map PeerId (Schedule ChainUpdate)
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
..}
    where
      genUpdateSchedule :: DiffusionPipeliningSupport -> Gen (Schedule ChainUpdate)
genUpdateSchedule DiffusionPipeliningSupport
diffusionPipelining =
          UpdateBehavior -> SecurityParam -> Int -> Gen [ChainUpdate]
genChainUpdates UpdateBehavior
behavior SecurityParam
maxRollback Int
20 Gen [ChainUpdate]
-> ([ChainUpdate] -> Gen (Schedule ChainUpdate))
-> Gen (Schedule ChainUpdate)
forall a b. Gen a -> (a -> Gen b) -> Gen b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [ChainUpdate] -> Gen (Schedule ChainUpdate)
forall a. [a] -> Gen (Schedule a)
genSchedule
        where
          behavior :: UpdateBehavior
behavior = case DiffusionPipeliningSupport
diffusionPipelining of
            DiffusionPipeliningSupport
DiffusionPipeliningOn  -> UpdateBehavior
TentativeChainBehavior
            DiffusionPipeliningSupport
DiffusionPipeliningOff -> UpdateBehavior
SelectedChainBehavior

      -- Only use a small k to avoid rolling forward by a big chain.
      maxRollback :: SecurityParam
maxRollback = Word64 -> SecurityParam
SecurityParam Word64
5

  shrink :: BlockFetchClientTestSetup -> [BlockFetchClientTestSetup]
shrink BlockFetchClientTestSetup{Map PeerId (Schedule ChainUpdate)
DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
peerUpdates :: BlockFetchClientTestSetup -> Map PeerId (Schedule ChainUpdate)
blockFetchMode :: BlockFetchClientTestSetup -> FetchMode
blockFetchCfg :: BlockFetchClientTestSetup -> BlockFetchConfiguration
blockFetchPipelining :: BlockFetchClientTestSetup -> DiffusionPipeliningSupport
peerUpdates :: Map PeerId (Schedule ChainUpdate)
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
..} =
      -- If we have multiple peers, check if removing the peer still
      -- yields an error
      [ BlockFetchClientTestSetup {
            peerUpdates :: Map PeerId (Schedule ChainUpdate)
peerUpdates = PeerId
-> Map PeerId (Schedule ChainUpdate)
-> Map PeerId (Schedule ChainUpdate)
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete PeerId
peerId Map PeerId (Schedule ChainUpdate)
peerUpdates
          , DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
..
          }
      | [PeerId] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [PeerId]
peerIds Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
1
      , PeerId
peerId <- [PeerId]
peerIds
      ] [BlockFetchClientTestSetup]
-> [BlockFetchClientTestSetup] -> [BlockFetchClientTestSetup]
forall a. Semigroup a => a -> a -> a
<>
      -- Shrink the schedules for all peers simultaneously
      [ BlockFetchClientTestSetup {
            peerUpdates :: Map PeerId (Schedule ChainUpdate)
peerUpdates = PeerId
-> Schedule ChainUpdate
-> Map PeerId (Schedule ChainUpdate)
-> Map PeerId (Schedule ChainUpdate)
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
peerId Schedule ChainUpdate
updates Map PeerId (Schedule ChainUpdate)
peerUpdates
          , DiffusionPipeliningSupport
FetchMode
BlockFetchConfiguration
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
blockFetchMode :: FetchMode
blockFetchCfg :: BlockFetchConfiguration
blockFetchPipelining :: DiffusionPipeliningSupport
..
          }
      | PeerId
peerId <- [PeerId]
peerIds
      , Schedule ChainUpdate
updates <-
          (Schedule ChainUpdate -> Bool)
-> [Schedule ChainUpdate] -> [Schedule ChainUpdate]
forall a. (a -> Bool) -> [a] -> [a]
filter (Bool -> Bool
not (Bool -> Bool)
-> (Schedule ChainUpdate -> Bool) -> Schedule ChainUpdate -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [ChainUpdate] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ([ChainUpdate] -> Bool)
-> (Schedule ChainUpdate -> [ChainUpdate])
-> Schedule ChainUpdate
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Schedule ChainUpdate -> [ChainUpdate]
forall a. Schedule a -> [a]
joinSchedule) ([Schedule ChainUpdate] -> [Schedule ChainUpdate])
-> [Schedule ChainUpdate] -> [Schedule ChainUpdate]
forall a b. (a -> b) -> a -> b
$
            Schedule ChainUpdate -> [Schedule ChainUpdate]
forall a. Schedule a -> [Schedule a]
shrinkSchedule (Map PeerId (Schedule ChainUpdate)
peerUpdates Map PeerId (Schedule ChainUpdate) -> PeerId -> Schedule ChainUpdate
forall k a. Ord k => Map k a -> k -> a
Map.! PeerId
peerId)
      ]
    where
      peerIds :: [PeerId]
peerIds = Map PeerId (Schedule ChainUpdate) -> [PeerId]
forall k a. Map k a -> [k]
Map.keys Map PeerId (Schedule ChainUpdate)
peerUpdates

newtype PeerId = PeerId Int
  deriving stock (Int -> PeerId -> String -> String
[PeerId] -> String -> String
PeerId -> String
(Int -> PeerId -> String -> String)
-> (PeerId -> String)
-> ([PeerId] -> String -> String)
-> Show PeerId
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> PeerId -> String -> String
showsPrec :: Int -> PeerId -> String -> String
$cshow :: PeerId -> String
show :: PeerId -> String
$cshowList :: [PeerId] -> String -> String
showList :: [PeerId] -> String -> String
Show, PeerId -> PeerId -> Bool
(PeerId -> PeerId -> Bool)
-> (PeerId -> PeerId -> Bool) -> Eq PeerId
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: PeerId -> PeerId -> Bool
== :: PeerId -> PeerId -> Bool
$c/= :: PeerId -> PeerId -> Bool
/= :: PeerId -> PeerId -> Bool
Eq, Eq PeerId
Eq PeerId =>
(PeerId -> PeerId -> Ordering)
-> (PeerId -> PeerId -> Bool)
-> (PeerId -> PeerId -> Bool)
-> (PeerId -> PeerId -> Bool)
-> (PeerId -> PeerId -> Bool)
-> (PeerId -> PeerId -> PeerId)
-> (PeerId -> PeerId -> PeerId)
-> Ord PeerId
PeerId -> PeerId -> Bool
PeerId -> PeerId -> Ordering
PeerId -> PeerId -> PeerId
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: PeerId -> PeerId -> Ordering
compare :: PeerId -> PeerId -> Ordering
$c< :: PeerId -> PeerId -> Bool
< :: PeerId -> PeerId -> Bool
$c<= :: PeerId -> PeerId -> Bool
<= :: PeerId -> PeerId -> Bool
$c> :: PeerId -> PeerId -> Bool
> :: PeerId -> PeerId -> Bool
$c>= :: PeerId -> PeerId -> Bool
>= :: PeerId -> PeerId -> Bool
$cmax :: PeerId -> PeerId -> PeerId
max :: PeerId -> PeerId -> PeerId
$cmin :: PeerId -> PeerId -> PeerId
min :: PeerId -> PeerId -> PeerId
Ord)
  deriving newtype (PeerId -> String
(PeerId -> String) -> Condense PeerId
forall a. (a -> String) -> Condense a
$ccondense :: PeerId -> String
condense :: PeerId -> String
Condense, Eq PeerId
Eq PeerId =>
(Int -> PeerId -> Int) -> (PeerId -> Int) -> Hashable PeerId
Int -> PeerId -> Int
PeerId -> Int
forall a. Eq a => (Int -> a -> Int) -> (a -> Int) -> Hashable a
$chashWithSalt :: Int -> PeerId -> Int
hashWithSalt :: Int -> PeerId -> Int
$chash :: PeerId -> Int
hash :: PeerId -> Int
Hashable, Int -> PeerId
PeerId -> Int
PeerId -> [PeerId]
PeerId -> PeerId
PeerId -> PeerId -> [PeerId]
PeerId -> PeerId -> PeerId -> [PeerId]
(PeerId -> PeerId)
-> (PeerId -> PeerId)
-> (Int -> PeerId)
-> (PeerId -> Int)
-> (PeerId -> [PeerId])
-> (PeerId -> PeerId -> [PeerId])
-> (PeerId -> PeerId -> [PeerId])
-> (PeerId -> PeerId -> PeerId -> [PeerId])
-> Enum PeerId
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: PeerId -> PeerId
succ :: PeerId -> PeerId
$cpred :: PeerId -> PeerId
pred :: PeerId -> PeerId
$ctoEnum :: Int -> PeerId
toEnum :: Int -> PeerId
$cfromEnum :: PeerId -> Int
fromEnum :: PeerId -> Int
$cenumFrom :: PeerId -> [PeerId]
enumFrom :: PeerId -> [PeerId]
$cenumFromThen :: PeerId -> PeerId -> [PeerId]
enumFromThen :: PeerId -> PeerId -> [PeerId]
$cenumFromTo :: PeerId -> PeerId -> [PeerId]
enumFromTo :: PeerId -> PeerId -> [PeerId]
$cenumFromThenTo :: PeerId -> PeerId -> PeerId -> [PeerId]
enumFromThenTo :: PeerId -> PeerId -> PeerId -> [PeerId]
Enum, PeerId
PeerId -> PeerId -> Bounded PeerId
forall a. a -> a -> Bounded a
$cminBound :: PeerId
minBound :: PeerId
$cmaxBound :: PeerId
maxBound :: PeerId
Bounded)

{-------------------------------------------------------------------------------
  Utilities
-------------------------------------------------------------------------------}

infiniteDelay :: MonadSTM m => m a
infiniteDelay :: forall (m :: * -> *) a. MonadSTM m => m a
infiniteDelay = STM m a -> m a
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m a
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry

chainToAnchoredFragment :: HasHeader blk => Chain blk -> AnchoredFragment blk
chainToAnchoredFragment :: forall blk. HasHeader blk => Chain blk -> AnchoredFragment blk
chainToAnchoredFragment =
    Anchor blk
-> [blk] -> AnchoredSeq (WithOrigin SlotNo) (Anchor blk) blk
forall v a b. Anchorable v a b => a -> [b] -> AnchoredSeq v a b
AF.fromOldestFirst Anchor blk
forall block. Anchor block
AF.AnchorGenesis ([blk] -> AnchoredSeq (WithOrigin SlotNo) (Anchor blk) blk)
-> (Chain blk -> [blk])
-> Chain blk
-> AnchoredSeq (WithOrigin SlotNo) (Anchor blk) blk
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Chain blk -> [blk]
forall block. Chain block -> [block]
Chain.toOldestFirst