{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE PolyKinds #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# LANGUAGE TypeApplications #-}
{-# LANGUAGE TypeFamilies #-}
{-# LANGUAGE UndecidableInstances #-}

-- | Setup network
module Test.ThreadNet.Network (
    CalcMessageDelay (..)
  , ForgeEbbEnv (..)
  , RekeyM
  , TestNodeInitialization (..)
  , ThreadNetworkArgs (..)
  , TracingConstraints
  , noCalcMessageDelay
  , plainTestNodeInitialization
  , runThreadNetwork
    -- * Tracers
  , MiniProtocolFatalException (..)
  , MiniProtocolState (..)
    -- * Test Output
  , NodeDBs (..)
  , NodeOutput (..)
  , TestOutput (..)
  ) where

import           Cardano.Network.PeerSelection.Bootstrap
                     (UseBootstrapPeers (..))
import           Codec.CBOR.Read (DeserialiseFailure)
import qualified Control.Concurrent.Class.MonadSTM as MonadSTM
import           Control.Concurrent.Class.MonadSTM.Strict (newTMVar)
import qualified Control.Exception as Exn
import           Control.Monad
import           Control.Monad.Class.MonadTime.SI (MonadTime)
import           Control.Monad.Class.MonadTimer.SI (MonadTimer)
import qualified Control.Monad.Except as Exc
import           Control.ResourceRegistry
import           Control.Tracer
import qualified Data.ByteString.Lazy as Lazy
import           Data.Functor.Contravariant ((>$<))
import           Data.Functor.Identity (Identity)
import qualified Data.List as List
import qualified Data.List.NonEmpty as NE
import           Data.Map.Strict (Map)
import qualified Data.Map.Strict as Map
import           Data.Set (Set)
import qualified Data.Set as Set
import qualified Data.Typeable as Typeable
import           Data.Void (Void)
import           GHC.Stack
import           Network.TypedProtocol.Codec (AnyMessage (..), CodecFailure,
                     mapFailureCodec)
import qualified Network.TypedProtocol.Codec as Codec
import           Ouroboros.Consensus.Block
import           Ouroboros.Consensus.BlockchainTime
import           Ouroboros.Consensus.Config
import           Ouroboros.Consensus.Ledger.Abstract
import           Ouroboros.Consensus.Ledger.Extended
import           Ouroboros.Consensus.Ledger.Inspect
import           Ouroboros.Consensus.Ledger.SupportsMempool
import           Ouroboros.Consensus.Ledger.SupportsProtocol
import           Ouroboros.Consensus.Ledger.Tables.Utils
import           Ouroboros.Consensus.Mempool
import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client as CSClient
import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client.HistoricityCheck as HistoricityCheck
import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck as InFutureCheck
import qualified Ouroboros.Consensus.Network.NodeToNode as NTN
import           Ouroboros.Consensus.Node.ExitPolicy
import           Ouroboros.Consensus.Node.Genesis
import qualified Ouroboros.Consensus.Node.GSM as GSM
import           Ouroboros.Consensus.Node.InitStorage
import           Ouroboros.Consensus.Node.NetworkProtocolVersion
import           Ouroboros.Consensus.Node.ProtocolInfo
import           Ouroboros.Consensus.Node.Run
import           Ouroboros.Consensus.Node.Tracers
import           Ouroboros.Consensus.NodeId
import           Ouroboros.Consensus.NodeKernel as NodeKernel
import           Ouroboros.Consensus.Protocol.Abstract
import qualified Ouroboros.Consensus.Storage.ChainDB as ChainDB
import qualified Ouroboros.Consensus.Storage.ChainDB.API.Types.InvalidBlockPunishment as InvalidBlockPunishment
import           Ouroboros.Consensus.Storage.ChainDB.Impl.Args
import           Ouroboros.Consensus.Storage.ChainDB.Impl.Types
import qualified Ouroboros.Consensus.Storage.ImmutableDB as ImmutableDB
import           Ouroboros.Consensus.Storage.LedgerDB
import           Ouroboros.Consensus.Storage.LedgerDB as LedgerDB
import qualified Ouroboros.Consensus.Storage.VolatileDB as VolatileDB
import           Ouroboros.Consensus.Util.Assert
import           Ouroboros.Consensus.Util.Condense
import           Ouroboros.Consensus.Util.Enclose (pattern FallingEdge)
import           Ouroboros.Consensus.Util.IOLike
import           Ouroboros.Consensus.Util.Orphans ()
import           Ouroboros.Consensus.Util.RedundantConstraints
import           Ouroboros.Consensus.Util.STM
import           Ouroboros.Consensus.Util.Time
import qualified Ouroboros.Network.AnchoredFragment as AF
import           Ouroboros.Network.BlockFetch (BlockFetchConfiguration (..),
                     TraceLabelPeer (..))
import           Ouroboros.Network.Channel
import           Ouroboros.Network.ControlMessage (ControlMessage (..))
import           Ouroboros.Network.Mock.Chain (Chain (Genesis))
import           Ouroboros.Network.NodeToNode (ConnectionId (..),
                     ExpandedInitiatorContext (..), IsBigLedgerPeer (..),
                     MiniProtocolParameters (..), ResponderContext (..))
import           Ouroboros.Network.PeerSelection.Governor
                     (makePublicPeerSelectionStateVar)
import           Ouroboros.Network.PeerSelection.PeerMetric (nullMetric)
import           Ouroboros.Network.Point (WithOrigin (..))
import qualified Ouroboros.Network.Protocol.ChainSync.Type as CS
import           Ouroboros.Network.Protocol.KeepAlive.Type
import           Ouroboros.Network.Protocol.Limits (waitForever)
import           Ouroboros.Network.Protocol.LocalStateQuery.Type
import           Ouroboros.Network.Protocol.PeerSharing.Type (PeerSharing)
import           Ouroboros.Network.Protocol.TxSubmission2.Type
import qualified System.FS.Sim.MockFS as Mock
import           System.FS.Sim.MockFS (MockFS)
import           System.Random (mkStdGen, split)
import           Test.ThreadNet.TxGen
import           Test.ThreadNet.Util.NodeJoinPlan
import           Test.ThreadNet.Util.NodeRestarts
import           Test.ThreadNet.Util.NodeTopology
import           Test.ThreadNet.Util.Seed
import           Test.Util.ChainDB
import qualified Test.Util.HardFork.Future as HFF
import           Test.Util.HardFork.Future (Future)
import qualified Test.Util.HardFork.OracularClock as OracularClock
import           Test.Util.HardFork.OracularClock (OracularClock (..))
import           Test.Util.Slots (NumSlots (..))
import           Test.Util.Time
import           Test.Util.Tracer

-- | How to forge an EBB
--
data ForgeEbbEnv blk = ForgeEbbEnv
  { forall blk.
ForgeEbbEnv blk
-> TopLevelConfig blk -> SlotNo -> BlockNo -> ChainHash blk -> blk
forgeEBB ::
         TopLevelConfig blk
      -> SlotNo
         -- EBB slot
      -> BlockNo
         -- EBB block number (i.e. that of its predecessor)
      -> ChainHash blk
         -- EBB predecessor's hash
      -> blk
  }
instance Show (ForgeEbbEnv blk) where
  showsPrec :: Int -> ForgeEbbEnv blk -> ShowS
showsPrec Int
p ForgeEbbEnv blk
_ = Bool -> ShowS -> ShowS
showParen (Int
p Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
10) (ShowS -> ShowS) -> ShowS -> ShowS
forall a b. (a -> b) -> a -> b
$ String -> ShowS
showString String
"ForgeEbbEnv _"

-- | How to rekey a node with a fresh operational key
--
-- When there is a 'NodeRekey' scheduled in the 'NodeRestarts', the test node
-- will restart and use 'tnaRekeyM' to compute its new 'ProtocolInfo'.
type RekeyM m blk =
     CoreNodeId
  -> ProtocolInfo blk
  -> m [BlockForging m blk]
  -> SlotNo
     -- ^ The slot in which the node is rekeying
  -> (SlotNo -> m EpochNo)
     -- ^ Which epoch the slot is in
  -> m (TestNodeInitialization m blk)
     -- ^ 'tniProtocolInfo' should include new delegation cert/operational key,
     -- and 'tniCrucialTxs' should include the new delegation certificate
     -- transaction

-- | Data used when starting/restarting a node
data TestNodeInitialization m blk = TestNodeInitialization
  { forall (m :: * -> *) blk.
TestNodeInitialization m blk -> [GenTx blk]
tniCrucialTxs   :: [GenTx blk]
    -- ^ these transactions are added immediately and repeatedly (whenever the
    -- 'ledgerTipSlot' changes)
    --
    -- In particular, a leading node's crucial transactions must (if valid)
    -- enter its mempool each slot /before/ the node takes the mempool snapshot
    -- that determines which transactions will be included in the block it's
    -- about to forge.
  , forall (m :: * -> *) blk.
TestNodeInitialization m blk -> ProtocolInfo blk
tniProtocolInfo :: ProtocolInfo blk
  , forall (m :: * -> *) blk.
TestNodeInitialization m blk -> m [BlockForging m blk]
tniBlockForging :: m [BlockForging m blk]
  }

plainTestNodeInitialization ::
     ProtocolInfo blk
  -> m [BlockForging m blk]
  -> TestNodeInitialization m blk
plainTestNodeInitialization :: forall blk (m :: * -> *).
ProtocolInfo blk
-> m [BlockForging m blk] -> TestNodeInitialization m blk
plainTestNodeInitialization ProtocolInfo blk
pInfo m [BlockForging m blk]
blockForging = TestNodeInitialization
    { tniCrucialTxs :: [GenTx blk]
tniCrucialTxs   = []
    , tniProtocolInfo :: ProtocolInfo blk
tniProtocolInfo = ProtocolInfo blk
pInfo
    , tniBlockForging :: m [BlockForging m blk]
tniBlockForging = m [BlockForging m blk]
blockForging
    }

-- | Compute the chain diffusion delay
--
-- This is the number of slots a 'CS.MsgRollForward' should arrive after the
-- forge slot of the header it is carrying.
--
-- It may depend on the @(sender, recipient)@, the current slot, and header.
newtype CalcMessageDelay blk = CalcMessageDelay
    ((CoreNodeId, CoreNodeId) -> SlotNo -> Header blk -> NumSlots)

noCalcMessageDelay :: CalcMessageDelay blk
noCalcMessageDelay :: forall blk. CalcMessageDelay blk
noCalcMessageDelay = ((CoreNodeId, CoreNodeId) -> SlotNo -> Header blk -> NumSlots)
-> CalcMessageDelay blk
forall blk.
((CoreNodeId, CoreNodeId) -> SlotNo -> Header blk -> NumSlots)
-> CalcMessageDelay blk
CalcMessageDelay (((CoreNodeId, CoreNodeId) -> SlotNo -> Header blk -> NumSlots)
 -> CalcMessageDelay blk)
-> ((CoreNodeId, CoreNodeId) -> SlotNo -> Header blk -> NumSlots)
-> CalcMessageDelay blk
forall a b. (a -> b) -> a -> b
$ \(CoreNodeId, CoreNodeId)
_ SlotNo
_ Header blk
_ -> Word64 -> NumSlots
NumSlots Word64
0

-- | This type occurs in records where most of the fields are data
instance Show (CalcMessageDelay blk) where
  show :: CalcMessageDelay blk -> String
show CalcMessageDelay blk
_ = String
"_CalcMessageDelay"

-- | Parameters for the test node net
--
data ThreadNetworkArgs m blk = ThreadNetworkArgs
  { forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> Maybe (ForgeEbbEnv blk)
tnaForgeEbbEnv  :: Maybe (ForgeEbbEnv blk)
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> Future
tnaFuture       :: Future
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NodeJoinPlan
tnaJoinPlan     :: NodeJoinPlan
  , forall (m :: * -> *) blk.
ThreadNetworkArgs m blk
-> CoreNodeId -> TestNodeInitialization m blk
tnaNodeInfo     :: CoreNodeId -> TestNodeInitialization m blk
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NumCoreNodes
tnaNumCoreNodes :: NumCoreNodes
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NumSlots
tnaNumSlots     :: NumSlots
  , forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> CalcMessageDelay blk
tnaMessageDelay :: CalcMessageDelay blk
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> Seed
tnaSeed         :: Seed
  , forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> Maybe (m (RekeyM m blk))
tnaMkRekeyM     :: Maybe (m (RekeyM m blk))
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NodeRestarts
tnaRestarts     :: NodeRestarts
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NodeTopology
tnaTopology     :: NodeTopology
  , forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> TxGenExtra blk
tnaTxGenExtra   :: TxGenExtra blk
  , forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> NodeToNodeVersion
tnaVersion      :: NodeToNodeVersion
  , forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> BlockNodeToNodeVersion blk
tnaBlockVersion :: BlockNodeToNodeVersion blk
  }

{-------------------------------------------------------------------------------
  Vertex and Edge Statuses
-------------------------------------------------------------------------------}

-- | A /vertex/ denotes the \"operator of a node\"; in production, that's
-- typically a person.
--
-- There is always exactly one vertex for each genesis key. When its current
-- node instance crashes/terminates, the vertex replaces it with a new one.
-- Every node instance created by a vertex uses the same file system.
--
-- The term \"vertex\" is only explicitly used in this module. However, the
-- concept exists throughout the project; it's usually denoted by the term
-- \"node\", which can mean either \"vertex\" or \"node instance\". We take
-- more care than usual in this module to be explicit, but still often rely on
-- context.
--
data VertexStatus m blk
  = VDown (Chain blk) (LedgerState blk EmptyMK)
    -- ^ The vertex does not currently have a node instance; its previous
    -- instance stopped with this chain and ledger state (empty/initial before
    -- first instance)
  | VFalling
    -- ^ The vertex has a node instance, but it is about to transition to
    -- 'VDown' as soon as its edges transition to 'EDown'.
  | VUp !(NodeKernel m NodeId Void blk) !(LimitedApp m NodeId blk)
    -- ^ The vertex currently has a node instance, with these handles.

-- | A directed /edge/ denotes the \"operator of a node-to-node connection\";
-- in production, that's generally the TCP connection and the networking layers
-- built atop it.
--
-- There are always exactly two edges between two vertices that are connected
-- by the 'NodeTopology': one for the client-server relationship in each
-- direction. When the mini protocol instances crash, the edge replaces them
-- with new instances, possibly after a delay (see 'RestartCause').
--
-- (We do not need 'EFalling' because node instances can exist without mini
-- protocols; we only need 'VFalling' because mini protocol instances cannot
-- exist without node instances.)
--
data EdgeStatus
  = EDown
    -- ^ The edge does not currently have mini protocol instances.
  | EUp
    -- ^ The edge currently has mini protocol instances.
  deriving (EdgeStatus -> EdgeStatus -> Bool
(EdgeStatus -> EdgeStatus -> Bool)
-> (EdgeStatus -> EdgeStatus -> Bool) -> Eq EdgeStatus
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: EdgeStatus -> EdgeStatus -> Bool
== :: EdgeStatus -> EdgeStatus -> Bool
$c/= :: EdgeStatus -> EdgeStatus -> Bool
/= :: EdgeStatus -> EdgeStatus -> Bool
Eq)

type VertexStatusVar m blk = StrictTVar m (VertexStatus m blk)
type EdgeStatusVar m = StrictTVar m EdgeStatus

{-------------------------------------------------------------------------------
  Running the node net
-------------------------------------------------------------------------------}

-- | Setup a network of core nodes, where each joins according to the node join
-- plan and is interconnected according to the node topology
--
-- We run for the specified number of blocks, then return the final state of
-- each node.
runThreadNetwork :: forall m blk.
                    ( IOLike m
                    , MonadTime m
                    , MonadTimer m
                    , RunNode blk
                    , TxGen blk
                    , TracingConstraints blk
                    , HasCallStack
                    )
                 => SystemTime m -> ThreadNetworkArgs m blk -> m (TestOutput blk)
runThreadNetwork :: forall (m :: * -> *) blk.
(IOLike m, MonadTime m, MonadTimer m, RunNode blk, TxGen blk,
 TracingConstraints blk, HasCallStack) =>
SystemTime m -> ThreadNetworkArgs m blk -> m (TestOutput blk)
runThreadNetwork SystemTime m
systemTime ThreadNetworkArgs
  { tnaForgeEbbEnv :: forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> Maybe (ForgeEbbEnv blk)
tnaForgeEbbEnv    = Maybe (ForgeEbbEnv blk)
mbForgeEbbEnv
  , tnaFuture :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> Future
tnaFuture         = Future
future
  , tnaJoinPlan :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NodeJoinPlan
tnaJoinPlan       = NodeJoinPlan
nodeJoinPlan
  , tnaNodeInfo :: forall (m :: * -> *) blk.
ThreadNetworkArgs m blk
-> CoreNodeId -> TestNodeInitialization m blk
tnaNodeInfo       = CoreNodeId -> TestNodeInitialization m blk
mkProtocolInfo
  , tnaNumCoreNodes :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NumCoreNodes
tnaNumCoreNodes   = NumCoreNodes
numCoreNodes
  , tnaNumSlots :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NumSlots
tnaNumSlots       = NumSlots
numSlots
  , tnaMessageDelay :: forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> CalcMessageDelay blk
tnaMessageDelay   = CalcMessageDelay blk
calcMessageDelay
  , tnaSeed :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> Seed
tnaSeed           = Seed
seed
  , tnaMkRekeyM :: forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> Maybe (m (RekeyM m blk))
tnaMkRekeyM       = Maybe (m (RekeyM m blk))
mbMkRekeyM
  , tnaRestarts :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NodeRestarts
tnaRestarts       = NodeRestarts
nodeRestarts
  , tnaTopology :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> NodeTopology
tnaTopology       = NodeTopology
nodeTopology
  , tnaTxGenExtra :: forall (m :: * -> *) blk. ThreadNetworkArgs m blk -> TxGenExtra blk
tnaTxGenExtra     = TxGenExtra blk
txGenExtra
  , tnaVersion :: forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> NodeToNodeVersion
tnaVersion        = NodeToNodeVersion
version
  , tnaBlockVersion :: forall (m :: * -> *) blk.
ThreadNetworkArgs m blk -> BlockNodeToNodeVersion blk
tnaBlockVersion   = BlockNodeToNodeVersion blk
blockVersion
  } = (ResourceRegistry m -> m (TestOutput blk)) -> m (TestOutput blk)
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
withRegistry ((ResourceRegistry m -> m (TestOutput blk)) -> m (TestOutput blk))
-> (ResourceRegistry m -> m (TestOutput blk)) -> m (TestOutput blk)
forall a b. (a -> b) -> a -> b
$ \ResourceRegistry m
sharedRegistry -> do
    mbRekeyM <- Maybe (m (RekeyM m blk)) -> m (Maybe (RekeyM m blk))
forall (t :: * -> *) (m :: * -> *) a.
(Traversable t, Monad m) =>
t (m a) -> m (t a)
forall (m :: * -> *) a. Monad m => Maybe (m a) -> m (Maybe a)
sequence Maybe (m (RekeyM m blk))
mbMkRekeyM

    -- This shared registry is used for 'newTestBlockchainTime' and the
    -- network communication threads. Each node will create its own registry
    -- for its ChainDB.
    -- TODO each node should run in its own thread and have its own (single
    -- top-level, bracketed) registry used to spawn all of the node's threads,
    -- including its own BlockchainTime. This will allow us to use
    -- ChainDB.withDB and avoid issues with termination and using registries
    -- from the wrong thread. To stop the network, wait for all the nodes'
    -- blockchain times to be done and then kill the main thread of each node,
    -- which should terminate all other threads it spawned.
    let clock = SystemTime m -> NumSlots -> Future -> OracularClock m
forall (m :: * -> *).
IOLike m =>
SystemTime m -> NumSlots -> Future -> OracularClock m
OracularClock.mkOracularClock SystemTime m
systemTime NumSlots
numSlots Future
future
    -- This function is organized around the notion of a network of nodes as a
    -- simple graph with no loops. The graph topology is determined by
    -- @nodeTopology@.
    --
    -- Each graph vertex is a node operator, and maintains its own Ouroboros
    -- core node, which in turn has its own private threads managing its
    -- internal state. Some nodes join the network later than others, according
    -- to @nodeJoinPlan@.
    --
    -- Each undirected edge denotes two opposing directed edges. Each directed
    -- edge denotes a bundle of mini protocols with client threads on the tail
    -- node and server threads on the head node. These mini protocols begin as
    -- soon as both nodes have joined the network, according to @nodeJoinPlan@.

    -- allocate the status variable for each vertex
    vertexStatusVars <- fmap Map.fromList $ do
      forM coreNodeIds $ \CoreNodeId
nid -> do
        -- assume they all start with the empty chain and the same initial
        -- ledger
        let nodeInitData :: TestNodeInitialization m blk
nodeInitData = CoreNodeId -> TestNodeInitialization m blk
mkProtocolInfo (Word64 -> CoreNodeId
CoreNodeId Word64
0)
            TestNodeInitialization{ProtocolInfo blk
tniProtocolInfo :: forall (m :: * -> *) blk.
TestNodeInitialization m blk -> ProtocolInfo blk
tniProtocolInfo :: ProtocolInfo blk
tniProtocolInfo} = TestNodeInitialization m blk
nodeInitData
            ProtocolInfo{ExtLedgerState blk ValuesMK
pInfoInitLedger :: ExtLedgerState blk ValuesMK
pInfoInitLedger :: forall b. ProtocolInfo b -> ExtLedgerState b ValuesMK
pInfoInitLedger} = ProtocolInfo blk
tniProtocolInfo
            ExtLedgerState{LedgerState blk ValuesMK
ledgerState :: LedgerState blk ValuesMK
ledgerState :: forall blk (mk :: MapKind).
ExtLedgerState blk mk -> LedgerState blk mk
ledgerState} = ExtLedgerState blk ValuesMK
pInfoInitLedger
        v <-
            VertexStatus m blk -> m (StrictTVar m (VertexStatus m blk))
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
uncheckedNewTVarM
          (VertexStatus m blk -> m (StrictTVar m (VertexStatus m blk)))
-> VertexStatus m blk -> m (StrictTVar m (VertexStatus m blk))
forall a b. (a -> b) -> a -> b
$ Chain blk -> LedgerState blk EmptyMK -> VertexStatus m blk
forall (m :: * -> *) blk.
Chain blk -> LedgerState blk EmptyMK -> VertexStatus m blk
VDown Chain blk
forall block. Chain block
Genesis
          (LedgerState blk EmptyMK -> VertexStatus m blk)
-> LedgerState blk EmptyMK -> VertexStatus m blk
forall a b. (a -> b) -> a -> b
$ LedgerState blk ValuesMK -> LedgerState blk EmptyMK
forall (l :: LedgerStateKind) (mk :: MapKind).
HasLedgerTables l =>
l mk -> l EmptyMK
forgetLedgerTables LedgerState blk ValuesMK
ledgerState
        pure (nid, v)

    -- fork the directed edges, which also allocates their status variables
    let uedges = NodeTopology -> [(CoreNodeId, CoreNodeId)]
edgesNodeTopology NodeTopology
nodeTopology
    edgeStatusVars <- fmap (Map.fromList . concat) $ do
      -- assume they all use the same CodecConfig
      let nodeInitData = CoreNodeId -> TestNodeInitialization m blk
mkProtocolInfo (Word64 -> CoreNodeId
CoreNodeId Word64
0)
          TestNodeInitialization{tniProtocolInfo} = nodeInitData
          ProtocolInfo{pInfoConfig} = tniProtocolInfo
          codecConfig = TopLevelConfig blk -> CodecConfig blk
forall blk. TopLevelConfig blk -> CodecConfig blk
configCodec TopLevelConfig blk
pInfoConfig
      forM uedges $ \(CoreNodeId, CoreNodeId)
uedge -> do
        ResourceRegistry m
-> OracularClock m
-> Tracer m (SlotNo, MiniProtocolState)
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> Map CoreNodeId (StrictTVar m (VertexStatus m blk))
-> (CoreNodeId, CoreNodeId)
-> m [((CoreNodeId, CoreNodeId), EdgeStatusVar m)]
forall (m :: * -> *) blk.
(IOLike m, RunNode blk, HasCallStack) =>
ResourceRegistry m
-> OracularClock m
-> Tracer m (SlotNo, MiniProtocolState)
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> Map CoreNodeId (VertexStatusVar m blk)
-> (CoreNodeId, CoreNodeId)
-> m [((CoreNodeId, CoreNodeId), EdgeStatusVar m)]
forkBothEdges
          ResourceRegistry m
sharedRegistry
          OracularClock m
clock
          -- traces when/why the mini protocol instances start and stop
          Tracer m (SlotNo, MiniProtocolState)
forall (m :: * -> *) a. (Applicative m, Show a) => Tracer m a
nullDebugTracer
          (NodeToNodeVersion
version, BlockNodeToNodeVersion blk
blockVersion)
          (CodecConfig blk
codecConfig, CalcMessageDelay blk
calcMessageDelay)
          Map CoreNodeId (StrictTVar m (VertexStatus m blk))
vertexStatusVars
          (CoreNodeId, CoreNodeId)
uedge

    -- fork the vertices
    let nodesByJoinSlot =
          ((SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))
 -> SlotNo)
-> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
-> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
forall b a. Ord b => (a -> b) -> [a] -> [a]
List.sortOn (SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk))) -> SlotNo
forall a b. (a, b) -> a
fst ([(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
 -> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))])
-> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
-> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
forall a b. (a -> b) -> a -> b
$   -- sort non-descending by join slot
          ((CoreNodeId, StrictTVar m (VertexStatus m blk))
 -> (SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk))))
-> [(CoreNodeId, StrictTVar m (VertexStatus m blk))]
-> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
forall a b. (a -> b) -> [a] -> [b]
map (\nv :: (CoreNodeId, StrictTVar m (VertexStatus m blk))
nv@(CoreNodeId
n, StrictTVar m (VertexStatus m blk)
_) -> (CoreNodeId -> SlotNo
joinSlotOf CoreNodeId
n, (CoreNodeId, StrictTVar m (VertexStatus m blk))
nv)) ([(CoreNodeId, StrictTVar m (VertexStatus m blk))]
 -> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))])
-> [(CoreNodeId, StrictTVar m (VertexStatus m blk))]
-> [(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))]
forall a b. (a -> b) -> a -> b
$
          Map CoreNodeId (StrictTVar m (VertexStatus m blk))
-> [(CoreNodeId, StrictTVar m (VertexStatus m blk))]
forall k a. Map k a -> [(k, a)]
Map.toList Map CoreNodeId (StrictTVar m (VertexStatus m blk))
vertexStatusVars
    vertexInfos0 <- forM nodesByJoinSlot $ \(SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))
vertexData -> do
      let (SlotNo
joinSlot, (CoreNodeId
coreNodeId, StrictTVar m (VertexStatus m blk)
vertexStatusVar)) = (SlotNo, (CoreNodeId, StrictTVar m (VertexStatus m blk)))
vertexData

      -- the vertex cannot create its first node instance until the
      -- 'NodeJoinPlan' allows
      tooLate <- OracularClock m -> SlotNo -> m Bool
forall (m :: * -> *). OracularClock m -> SlotNo -> m Bool
OracularClock.blockUntilSlot OracularClock m
clock SlotNo
joinSlot
      when tooLate $ do
        error $ "unsatisfiable nodeJoinPlan: " ++ show coreNodeId

      -- fork the per-vertex state variables, including the mock filesystem
      (nodeInfo, readNodeInfo) <- newNodeInfo

      -- a tvar containing the next slot to be recorded in
      -- nodeEventsTipBlockNos
      nextInstrSlotVar <- uncheckedNewTVarM joinSlot

      let myEdgeStatusVars =
            [ EdgeStatusVar m
v
            | ((CoreNodeId
n1, CoreNodeId
n2), EdgeStatusVar m
v) <- Map (CoreNodeId, CoreNodeId) (EdgeStatusVar m)
-> [((CoreNodeId, CoreNodeId), EdgeStatusVar m)]
forall k a. Map k a -> [(k, a)]
Map.toList Map (CoreNodeId, CoreNodeId) (EdgeStatusVar m)
edgeStatusVars
            , CoreNodeId
coreNodeId CoreNodeId -> [CoreNodeId] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [CoreNodeId
n1, CoreNodeId
n2]
            ]
      forkVertex
        mbRekeyM
        clock
        joinSlot
        sharedRegistry
        coreNodeId
        vertexStatusVar
        myEdgeStatusVars
        nodeInfo
        nextInstrSlotVar

      forkInstrumentation
        clock
        sharedRegistry
        vertexStatusVar
        nodeInfo
        nextInstrSlotVar

      return (coreNodeId, vertexStatusVar, readNodeInfo)

    -- Wait for the last slot to end
    OracularClock.waitUntilDone clock

    -- Collect all nodes' final chains
    vertexInfos <-
      atomically $
      forM vertexInfos0 $ \(CoreNodeId
coreNodeId, StrictTVar m (VertexStatus m blk)
vertexStatusVar, m (NodeInfo blk MockFS [])
readNodeInfo) -> do
        StrictTVar m (VertexStatus m blk) -> STM m (VertexStatus m blk)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (VertexStatus m blk)
vertexStatusVar STM m (VertexStatus m blk)
-> (VertexStatus m blk
    -> STM
         m
         (CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
          LedgerState blk EmptyMK))
-> STM
     m
     (CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
      LedgerState blk EmptyMK)
forall a b. STM m a -> (a -> STM m b) -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          VDown Chain blk
ch LedgerState blk EmptyMK
ldgr -> (CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
 LedgerState blk EmptyMK)
-> STM
     m
     (CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
      LedgerState blk EmptyMK)
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (CoreNodeId
coreNodeId, m (NodeInfo blk MockFS [])
readNodeInfo, Chain blk
ch, LedgerState blk EmptyMK
ldgr)
          VertexStatus m blk
_             -> STM
  m
  (CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
   LedgerState blk EmptyMK)
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry

    mkTestOutput vertexInfos
  where
    ()
_ = Proxy (Show (LedgerView (BlockProtocol blk))) -> ()
forall (c :: Constraint) (proxy :: Constraint -> *).
c =>
proxy c -> ()
keepRedundantConstraint (forall {k} (t :: k). Proxy t
forall (t :: Constraint). Proxy t
Proxy @(Show (LedgerView (BlockProtocol blk))))

    -- epoch size of the first era (ie the one that might have EBBs)
    epochSize0 :: EpochSize
    epochSize0 :: EpochSize
epochSize0 = Future -> EpochSize
HFF.futureFirstEpochSize Future
future

    coreNodeIds :: [CoreNodeId]
    coreNodeIds :: [CoreNodeId]
coreNodeIds = NumCoreNodes -> [CoreNodeId]
enumCoreNodes NumCoreNodes
numCoreNodes

    joinSlotOf :: CoreNodeId -> SlotNo
    joinSlotOf :: CoreNodeId -> SlotNo
joinSlotOf = HasCallStack => NodeJoinPlan -> CoreNodeId -> SlotNo
NodeJoinPlan -> CoreNodeId -> SlotNo
coreNodeIdJoinSlot NodeJoinPlan
nodeJoinPlan

    forkVertex
      :: Maybe (RekeyM m blk)
      -> OracularClock m
      -> SlotNo
      -> ResourceRegistry m
      -> CoreNodeId
      -> VertexStatusVar m blk
      -> [EdgeStatusVar m]
      -> NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
      -> StrictTVar m SlotNo
      -> m ()
    forkVertex :: Maybe (RekeyM m blk)
-> OracularClock m
-> SlotNo
-> ResourceRegistry m
-> CoreNodeId
-> StrictTVar m (VertexStatus m blk)
-> [EdgeStatusVar m]
-> NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
-> StrictTVar m SlotNo
-> m ()
forkVertex
      Maybe (RekeyM m blk)
mbRekeyM
      OracularClock m
clock
      SlotNo
joinSlot
      ResourceRegistry m
sharedRegistry
      CoreNodeId
coreNodeId
      StrictTVar m (VertexStatus m blk)
vertexStatusVar
      [EdgeStatusVar m]
edgeStatusVars
      NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
nodeInfo
      StrictTVar m SlotNo
nextInstrSlotVar =
        m (Thread m ()) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (Thread m ()) -> m ()) -> m (Thread m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ ResourceRegistry m -> String -> m () -> m (Thread m ())
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, HasCallStack) =>
ResourceRegistry m -> String -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry m
sharedRegistry String
label (m () -> m (Thread m ())) -> m () -> m (Thread m ())
forall a b. (a -> b) -> a -> b
$ do
          SlotNo
-> ProtocolInfo blk
-> m [BlockForging m blk]
-> NodeRestart
-> Map SlotNo NodeRestart
-> m ()
loop SlotNo
0 ProtocolInfo blk
tniProtocolInfo m [BlockForging m blk]
tniBlockForging NodeRestart
NodeRestart Map SlotNo NodeRestart
restarts0
      where
        label :: String
label = String
"vertex-" String -> ShowS
forall a. Semigroup a => a -> a -> a
<> CoreNodeId -> String
forall a. Condense a => a -> String
condense CoreNodeId
coreNodeId

        TestNodeInitialization
           { [GenTx blk]
tniCrucialTxs :: forall (m :: * -> *) blk.
TestNodeInitialization m blk -> [GenTx blk]
tniCrucialTxs :: [GenTx blk]
tniCrucialTxs
           , ProtocolInfo blk
tniProtocolInfo :: forall (m :: * -> *) blk.
TestNodeInitialization m blk -> ProtocolInfo blk
tniProtocolInfo :: ProtocolInfo blk
tniProtocolInfo
           , m [BlockForging m blk]
tniBlockForging :: forall (m :: * -> *) blk.
TestNodeInitialization m blk -> m [BlockForging m blk]
tniBlockForging :: m [BlockForging m blk]
tniBlockForging
           } = CoreNodeId -> TestNodeInitialization m blk
mkProtocolInfo CoreNodeId
coreNodeId

        restarts0 :: Map SlotNo NodeRestart
        restarts0 :: Map SlotNo NodeRestart
restarts0 = (Map CoreNodeId NodeRestart -> Maybe NodeRestart)
-> Map SlotNo (Map CoreNodeId NodeRestart)
-> Map SlotNo NodeRestart
forall a b k. (a -> Maybe b) -> Map k a -> Map k b
Map.mapMaybe (CoreNodeId -> Map CoreNodeId NodeRestart -> Maybe NodeRestart
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup CoreNodeId
coreNodeId) Map SlotNo (Map CoreNodeId NodeRestart)
m
          where
            NodeRestarts Map SlotNo (Map CoreNodeId NodeRestart)
m = NodeRestarts
nodeRestarts

        loop :: SlotNo
             -> ProtocolInfo blk
             -> m [BlockForging m blk]
             -> NodeRestart
             -> Map SlotNo NodeRestart -> m ()
        loop :: SlotNo
-> ProtocolInfo blk
-> m [BlockForging m blk]
-> NodeRestart
-> Map SlotNo NodeRestart
-> m ()
loop SlotNo
s ProtocolInfo blk
pInfo m [BlockForging m blk]
blockForging NodeRestart
nr Map SlotNo NodeRestart
rs = do
          -- a registry solely for the resources of this specific node instance
          (again, finalChain, finalLdgr) <- (ResourceRegistry m
 -> m (Maybe
         (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
          Map SlotNo NodeRestart),
       Chain blk, LedgerState blk EmptyMK))
-> m (Maybe
        (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
         Map SlotNo NodeRestart),
      Chain blk, LedgerState blk EmptyMK)
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
withRegistry ((ResourceRegistry m
  -> m (Maybe
          (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
           Map SlotNo NodeRestart),
        Chain blk, LedgerState blk EmptyMK))
 -> m (Maybe
         (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
          Map SlotNo NodeRestart),
       Chain blk, LedgerState blk EmptyMK))
-> (ResourceRegistry m
    -> m (Maybe
            (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
             Map SlotNo NodeRestart),
          Chain blk, LedgerState blk EmptyMK))
-> m (Maybe
        (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
         Map SlotNo NodeRestart),
      Chain blk, LedgerState blk EmptyMK)
forall a b. (a -> b) -> a -> b
$ \ResourceRegistry m
nodeRegistry -> do
            -- change the node's key and prepare a delegation transaction if
            -- the node is restarting because it just rekeyed
            tni' <- case (NodeRestart
nr, Maybe (RekeyM m blk)
mbRekeyM) of
              (NodeRestart
NodeRekey, Just RekeyM m blk
rekeyM) -> do
                RekeyM m blk
rekeyM CoreNodeId
coreNodeId ProtocolInfo blk
pInfo m [BlockForging m blk]
blockForging SlotNo
s (EpochNo -> m EpochNo
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (EpochNo -> m EpochNo)
-> (SlotNo -> EpochNo) -> SlotNo -> m EpochNo
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Future -> SlotNo -> EpochNo
HFF.futureSlotToEpoch Future
future)
              (NodeRestart, Maybe (RekeyM m blk))
_                        ->
                  TestNodeInitialization m blk -> m (TestNodeInitialization m blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TestNodeInitialization m blk -> m (TestNodeInitialization m blk))
-> TestNodeInitialization m blk -> m (TestNodeInitialization m blk)
forall a b. (a -> b) -> a -> b
$ ProtocolInfo blk
-> m [BlockForging m blk] -> TestNodeInitialization m blk
forall blk (m :: * -> *).
ProtocolInfo blk
-> m [BlockForging m blk] -> TestNodeInitialization m blk
plainTestNodeInitialization ProtocolInfo blk
pInfo m [BlockForging m blk]
blockForging
            let TestNodeInitialization
                  { tniCrucialTxs   = crucialTxs'
                  , tniProtocolInfo = pInfo'
                  , tniBlockForging = blockForging'
                  } = tni'

            -- allocate the node's internal state and fork its internal threads
            -- (specifically not the communication threads running the Mini
            -- Protocols, like the ChainSync Client)
            (kernel, app) <- forkNode
              coreNodeId
              clock
              joinSlot
              nodeRegistry
              pInfo'
              blockForging'
              nodeInfo
              (crucialTxs' ++ tniCrucialTxs)
            atomically $ writeTVar vertexStatusVar $ VUp kernel app

            -- wait until this node instance should stop
            again <- case Map.minViewWithKey rs of
              -- end of test
              Maybe ((SlotNo, NodeRestart), Map SlotNo NodeRestart)
Nothing               -> do
                OracularClock m -> m ()
forall (m :: * -> *). OracularClock m -> m ()
OracularClock.waitUntilDone OracularClock m
clock
                Maybe
  (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
   Map SlotNo NodeRestart)
-> m (Maybe
        (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
         Map SlotNo NodeRestart))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe
  (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
   Map SlotNo NodeRestart)
forall a. Maybe a
Nothing
              -- onset of schedule restart slot
              Just ((SlotNo
s', NodeRestart
nr'), Map SlotNo NodeRestart
rs') -> do
                -- wait until the node should stop
                tooLate <- OracularClock m -> SlotNo -> m Bool
forall (m :: * -> *). OracularClock m -> SlotNo -> m Bool
OracularClock.blockUntilSlot OracularClock m
clock SlotNo
s'
                when tooLate $ do
                  error $ "unsatisfiable nodeRestarts: "
                    ++ show (coreNodeId, s')

                -- this synchronization prevents a race with the
                -- instrumentation thread: we it want to record the current
                -- block number at the start of the slot, before this vertex
                -- restarts the node
                atomically $ do
                  nextSlot <- readTVar nextInstrSlotVar
                  check $ nextSlot > s'

                pure $ Just (s', pInfo', blockForging', nr', rs')

            -- stop threads that depend on/stimulate the kernel
            atomically $ writeTVar vertexStatusVar VFalling
            forM_ edgeStatusVars $ \EdgeStatusVar m
edgeStatusVar -> 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
$ do
              EdgeStatusVar m -> STM m EdgeStatus
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar EdgeStatusVar m
edgeStatusVar STM m EdgeStatus -> (EdgeStatus -> STM m ()) -> STM m ()
forall a b. STM m a -> (a -> STM m b) -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM m ()
forall (m :: * -> *). MonadSTM m => Bool -> STM m ()
check (Bool -> STM m ())
-> (EdgeStatus -> Bool) -> EdgeStatus -> STM m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (EdgeStatus -> EdgeStatus -> Bool
forall a. Eq a => a -> a -> Bool
== EdgeStatus
EDown)

            -- assuming nothing else is changing it, read the final chain
            let chainDB = NodeKernel m NodeId Void blk -> ChainDB m blk
forall (m :: * -> *) addrNTN addrNTC blk.
NodeKernel m addrNTN addrNTC blk -> ChainDB m blk
getChainDB NodeKernel m NodeId Void blk
kernel
            ExtLedgerState{ledgerState} <- atomically $
              ChainDB.getCurrentLedger chainDB
            finalChain <- ChainDB.toChain chainDB


            pure (again, finalChain, ledgerState)
            -- end of the node's withRegistry

          atomically $ writeTVar vertexStatusVar $
            VDown finalChain finalLdgr

          case again of
            Maybe
  (SlotNo, ProtocolInfo blk, m [BlockForging m blk], NodeRestart,
   Map SlotNo NodeRestart)
Nothing                                    -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            Just (SlotNo
s', ProtocolInfo blk
pInfo', m [BlockForging m blk]
blockForging', NodeRestart
nr', Map SlotNo NodeRestart
rs') -> SlotNo
-> ProtocolInfo blk
-> m [BlockForging m blk]
-> NodeRestart
-> Map SlotNo NodeRestart
-> m ()
loop SlotNo
s' ProtocolInfo blk
pInfo' m [BlockForging m blk]
blockForging' NodeRestart
nr' Map SlotNo NodeRestart
rs'

    -- | Instrumentation: record the tip's block number at the onset of the
    -- slot.
    --
    -- With such a short transaction (read a few TVars) we assume this runs (1)
    -- before anything else in the slot and (2) once per slot.
    forkInstrumentation
      :: OracularClock m
      -> ResourceRegistry m
      -> VertexStatusVar m blk
      -> NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
      -> StrictTVar m SlotNo
      -> m ()
    forkInstrumentation :: OracularClock m
-> ResourceRegistry m
-> StrictTVar m (VertexStatus m blk)
-> NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
-> StrictTVar m SlotNo
-> m ()
forkInstrumentation
      OracularClock m
clock
      ResourceRegistry m
registry
      StrictTVar m (VertexStatus m blk)
vertexStatusVar
      NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
nodeInfo
      StrictTVar m SlotNo
nextInstrSlotVar =
        m (m ()) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (m ()) -> m ()) -> m (m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ ResourceRegistry m
-> OracularClock m -> String -> (SlotNo -> m ()) -> m (m ())
forall (m :: * -> *).
HasCallStack =>
ResourceRegistry m
-> OracularClock m -> String -> (SlotNo -> m ()) -> m (m ())
OracularClock.forkEachSlot ResourceRegistry m
registry OracularClock m
clock String
lbl ((SlotNo -> m ()) -> m (m ())) -> (SlotNo -> m ()) -> m (m ())
forall a b. (a -> b) -> a -> b
$ \SlotNo
s -> do
          bno <- STM m (WithOrigin BlockNo) -> m (WithOrigin BlockNo)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (WithOrigin BlockNo) -> m (WithOrigin BlockNo))
-> STM m (WithOrigin BlockNo) -> m (WithOrigin BlockNo)
forall a b. (a -> b) -> a -> b
$ StrictTVar m (VertexStatus m blk) -> STM m (VertexStatus m blk)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (VertexStatus m blk)
vertexStatusVar STM m (VertexStatus m blk)
-> (VertexStatus m blk -> STM m (WithOrigin BlockNo))
-> STM m (WithOrigin BlockNo)
forall a b. STM m a -> (a -> STM m b) -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            VUp NodeKernel m NodeId Void blk
kernel LimitedApp m NodeId blk
_ -> ChainDB m blk -> STM m (WithOrigin BlockNo)
forall (m :: * -> *) blk.
(Monad (STM m), HasHeader (Header blk)) =>
ChainDB m blk -> STM m (WithOrigin BlockNo)
ChainDB.getTipBlockNo (NodeKernel m NodeId Void blk -> ChainDB m blk
forall (m :: * -> *) addrNTN addrNTC blk.
NodeKernel m addrNTN addrNTC blk -> ChainDB m blk
getChainDB NodeKernel m NodeId Void blk
kernel)
            VertexStatus m blk
_            -> STM m (WithOrigin BlockNo)
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry
          traceWith nodeEventsTipBlockNos (s, bno)
          atomically $ modifyTVar nextInstrSlotVar $ max (succ s)
      where
        NodeInfo{NodeEvents blk (Tracer m)
nodeInfoEvents :: NodeEvents blk (Tracer m)
nodeInfoEvents :: forall blk db (ev :: * -> *).
NodeInfo blk db ev -> NodeEvents blk ev
nodeInfoEvents}          = NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
nodeInfo
        NodeEvents{Tracer m (SlotNo, WithOrigin BlockNo)
nodeEventsTipBlockNos :: Tracer m (SlotNo, WithOrigin BlockNo)
nodeEventsTipBlockNos :: forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, WithOrigin BlockNo)
nodeEventsTipBlockNos} = NodeEvents blk (Tracer m)
nodeInfoEvents
        lbl :: String
lbl                               = String
"instrumentation"

    -- | Persistently attempt to add the given transactions to the mempool
    -- every time the ledger slot changes, even if successful!
    --
    -- If we add the transaction and then the mempools discards it for some
    -- reason, this thread will add it again.
    --
    forkCrucialTxs
      :: HasCallStack
      => OracularClock m
      -> SlotNo
      -> ResourceRegistry m
      -> (SlotNo -> STM m ())
      -> STM m (Point blk)
      -> (ResourceRegistry m -> m (ReadOnlyForker' m blk))
      -> Mempool m blk
      -> [GenTx blk]
         -- ^ valid transactions the node should immediately propagate
      -> m ()
    forkCrucialTxs :: HasCallStack =>
OracularClock m
-> SlotNo
-> ResourceRegistry m
-> (SlotNo -> STM m ())
-> STM m (Point blk)
-> (ResourceRegistry m -> m (ReadOnlyForker' m blk))
-> Mempool m blk
-> [GenTx blk]
-> m ()
forkCrucialTxs OracularClock m
clock SlotNo
s0 ResourceRegistry m
registry SlotNo -> STM m ()
unblockForge STM m (Point blk)
getTipPoint ResourceRegistry m -> m (ReadOnlyForker' m blk)
mforker Mempool m blk
mempool [GenTx blk]
txs0 = do
      m (Thread m Any) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (Thread m Any) -> m ()) -> m (Thread m Any) -> m ()
forall a b. (a -> b) -> a -> b
$ 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
"crucialTxs" (m Any -> m (Thread m Any)) -> m Any -> m (Thread m Any)
forall a b. (a -> b) -> a -> b
$ (ResourceRegistry m -> m Any) -> m Any
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
withRegistry ((ResourceRegistry m -> m Any) -> m Any)
-> (ResourceRegistry m -> m Any) -> m Any
forall a b. (a -> b) -> a -> b
$ \ResourceRegistry m
reg -> do

        let loop :: (SlotNo, [TicketNo]) -> m Any
loop (SlotNo
slot, [TicketNo]
mempFp) = do
              forker <- ResourceRegistry m -> m (ReadOnlyForker' m blk)
mforker ResourceRegistry m
reg
              extLedger <- atomically $ roforkerGetLedgerState forker
              let ledger = ExtLedgerState blk EmptyMK -> LedgerState blk EmptyMK
forall blk (mk :: MapKind).
ExtLedgerState blk mk -> LedgerState blk mk
ledgerState ExtLedgerState blk EmptyMK
extLedger
              roforkerClose forker

              _ <- addTxs mempool txs0

              -- See 'unblockForge' in 'forkNode'
              atomically $ unblockForge slot

              let
                -- a clock tick might render a crucial transaction valid
                slotChanged = do
                  let slot' :: SlotNo
slot' = SlotNo -> SlotNo
forall a. Enum a => a -> a
succ SlotNo
slot
                  _ <- OracularClock m -> SlotNo -> m Bool
forall (m :: * -> *). OracularClock m -> SlotNo -> m Bool
OracularClock.blockUntilSlot OracularClock m
clock SlotNo
slot'
                  pure (slot', mempFp)

                -- a new tx (e.g. added by TxSubmission) might render a crucial
                -- transaction valid
                mempChanged = do
                  let prjTno :: (a, TicketNo, c) -> TicketNo
prjTno (a
_a, TicketNo
b, c
_c) = TicketNo
b :: TicketNo
                      getMemp :: STM m [TicketNo]
getMemp = (((Validated (GenTx blk), TicketNo, TxMeasure blk) -> TicketNo)
-> [(Validated (GenTx blk), TicketNo, TxMeasure blk)] -> [TicketNo]
forall a b. (a -> b) -> [a] -> [b]
map (Validated (GenTx blk), TicketNo, TxMeasure blk) -> TicketNo
forall {a} {c}. (a, TicketNo, c) -> TicketNo
prjTno ([(Validated (GenTx blk), TicketNo, TxMeasure blk)] -> [TicketNo])
-> (MempoolSnapshot blk
    -> [(Validated (GenTx blk), TicketNo, TxMeasure blk)])
-> MempoolSnapshot blk
-> [TicketNo]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. MempoolSnapshot blk
-> [(Validated (GenTx blk), TicketNo, TxMeasure blk)]
forall blk.
MempoolSnapshot blk
-> [(Validated (GenTx blk), TicketNo, TxMeasure blk)]
snapshotTxs) (MempoolSnapshot blk -> [TicketNo])
-> STM m (MempoolSnapshot blk) -> STM m [TicketNo]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Mempool m blk -> STM m (MempoolSnapshot blk)
forall (m :: * -> *) blk.
Mempool m blk -> STM m (MempoolSnapshot blk)
getSnapshot Mempool m blk
mempool
                  (mempFp', _) <- STM m ([TicketNo], [TicketNo]) -> m ([TicketNo], [TicketNo])
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m ([TicketNo], [TicketNo]) -> m ([TicketNo], [TicketNo]))
-> STM m ([TicketNo], [TicketNo]) -> m ([TicketNo], [TicketNo])
forall a b. (a -> b) -> a -> b
$ ([TicketNo] -> [TicketNo])
-> [TicketNo] -> STM m [TicketNo] -> STM m ([TicketNo], [TicketNo])
forall (m :: * -> *) a b.
(MonadSTM m, Eq b) =>
(a -> b) -> b -> STM m a -> STM m (a, b)
blockUntilChanged [TicketNo] -> [TicketNo]
forall a. a -> a
id [TicketNo]
mempFp STM m [TicketNo]
getMemp
                  pure (slot, mempFp')

                -- a new ledger state might render a crucial transaction valid
                ldgrChanged = do
                  _ <- STM m (Point blk, Point blk) -> m (Point blk, Point blk)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Point blk, Point blk) -> m (Point blk, Point blk))
-> STM m (Point blk, Point blk) -> m (Point blk, Point blk)
forall a b. (a -> b) -> a -> b
$ (Point blk -> Point blk)
-> Point blk -> STM m (Point blk) -> STM m (Point blk, Point blk)
forall (m :: * -> *) a b.
(MonadSTM m, Eq b) =>
(a -> b) -> b -> STM m a -> STM m (a, b)
blockUntilChanged Point blk -> Point blk
forall a. a -> a
id (LedgerState blk EmptyMK -> Point blk
forall blk (mk :: MapKind).
UpdateLedger blk =>
LedgerState blk mk -> Point blk
ledgerTipPoint LedgerState blk EmptyMK
ledger) STM m (Point blk)
getTipPoint
                  pure (slot, mempFp)

              -- wake up when any of those change
              --
              -- key observation: it's OK to add the crucial txs too often
              fps' <- fmap (either (either id id) id) $
                slotChanged `race` mempChanged `race` ldgrChanged

              -- avoid the race in which we wake up before the mempool's
              -- background thread wakes up by mimicking it before we do
              -- anything else
              void $ syncWithLedger mempool

              loop fps'
        (SlotNo, [TicketNo]) -> m Any
loop (SlotNo
s0, [])

    -- | Produce transactions every time the slot changes and submit them to
    -- the mempool.
    forkTxProducer :: HasCallStack
                   => CoreNodeId
                   -> ResourceRegistry m
                   -> OracularClock m
                   -> TopLevelConfig blk
                   -> Seed
                   -> (ResourceRegistry m -> m (ReadOnlyForker' m blk))
                      -- ^ How to get the current ledger state
                   -> Mempool m blk
                   -> m ()
    forkTxProducer :: HasCallStack =>
CoreNodeId
-> ResourceRegistry m
-> OracularClock m
-> TopLevelConfig blk
-> Seed
-> (ResourceRegistry m -> m (ReadOnlyForker' m blk))
-> Mempool m blk
-> m ()
forkTxProducer CoreNodeId
coreNodeId ResourceRegistry m
registry OracularClock m
clock TopLevelConfig blk
cfg Seed
nodeSeed ResourceRegistry m -> m (ReadOnlyForker' m blk)
mforker Mempool m blk
mempool =
        m (m ()) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (m ()) -> m ()) -> m (m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ ResourceRegistry m
-> OracularClock m -> String -> (SlotNo -> m ()) -> m (m ())
forall (m :: * -> *).
HasCallStack =>
ResourceRegistry m
-> OracularClock m -> String -> (SlotNo -> m ()) -> m (m ())
OracularClock.forkEachSlot ResourceRegistry m
registry OracularClock m
clock String
"txProducer" ((SlotNo -> m ()) -> m (m ())) -> (SlotNo -> m ()) -> m (m ())
forall a b. (a -> b) -> a -> b
$ \SlotNo
curSlotNo -> (ResourceRegistry m -> m ()) -> m ()
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
withRegistry ((ResourceRegistry m -> m ()) -> m ())
-> (ResourceRegistry m -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \ResourceRegistry m
reg -> do
          forker <- ResourceRegistry m -> m (ReadOnlyForker' m blk)
mforker ResourceRegistry m
reg
          emptySt' <- atomically $ roforkerGetLedgerState forker
          let emptySt      = ExtLedgerState blk EmptyMK
emptySt'
              doRangeQuery = ReadOnlyForker' m blk
-> RangeQueryPrevious (ExtLedgerState blk)
-> m (LedgerTables (ExtLedgerState blk) ValuesMK)
forall (m :: * -> *) (l :: LedgerStateKind) blk.
ReadOnlyForker m l blk
-> RangeQueryPrevious l -> m (LedgerTables l ValuesMK)
roforkerRangeReadTables ReadOnlyForker' m blk
forker
          fullLedgerSt <- fmap ledgerState $ do
                fullUTxO <- doRangeQuery NoPreviousQuery
                pure $! withLedgerTables emptySt fullUTxO
          roforkerClose forker
          -- Combine the node's seed with the current slot number, to make sure
          -- we generate different transactions in each slot.
          let txs = Seed -> Gen [GenTx blk] -> [GenTx blk]
forall a. Seed -> Gen a -> a
runGen
                  (Seed
nodeSeed Seed -> Word64 -> Seed
forall a. Integral a => Seed -> a -> Seed
`combineWith` SlotNo -> Word64
unSlotNo SlotNo
curSlotNo)
                  (CoreNodeId
-> NumCoreNodes
-> SlotNo
-> TopLevelConfig blk
-> TxGenExtra blk
-> LedgerState blk ValuesMK
-> Gen [GenTx blk]
forall blk.
TxGen blk =>
CoreNodeId
-> NumCoreNodes
-> SlotNo
-> TopLevelConfig blk
-> TxGenExtra blk
-> LedgerState blk ValuesMK
-> Gen [GenTx blk]
testGenTxs CoreNodeId
coreNodeId NumCoreNodes
numCoreNodes SlotNo
curSlotNo TopLevelConfig blk
cfg TxGenExtra blk
txGenExtra LedgerState blk ValuesMK
fullLedgerSt)
          void $ addTxs mempool txs

    mkArgs :: ResourceRegistry m
           -> TopLevelConfig blk
           -> ExtLedgerState blk ValuesMK
           -> Tracer m (RealPoint blk, ExtValidationError blk)
              -- ^ invalid block tracer
           -> Tracer m (RealPoint blk, BlockNo)
              -- ^ added block tracer
           -> Tracer m (RealPoint blk, BlockNo)
              -- ^ block selection tracer
           -> Tracer m (LedgerUpdate blk)
              -- ^ ledger updates tracer
           -> Tracer m (ChainDB.TracePipeliningEvent blk)
           -> NodeDBs (StrictTMVar m MockFS)
           -> CoreNodeId
           -> ChainDbArgs Identity m blk
    mkArgs :: ResourceRegistry m
-> TopLevelConfig blk
-> ExtLedgerState blk ValuesMK
-> Tracer m (RealPoint blk, ExtValidationError blk)
-> Tracer m (RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
-> Tracer m (LedgerUpdate blk)
-> Tracer m (TracePipeliningEvent blk)
-> NodeDBs (StrictTMVar m MockFS)
-> CoreNodeId
-> ChainDbArgs Identity m blk
mkArgs
      ResourceRegistry m
registry
      TopLevelConfig blk
cfg ExtLedgerState blk ValuesMK
initLedger
      Tracer m (RealPoint blk, ExtValidationError blk)
invalidTracer Tracer m (RealPoint blk, BlockNo)
addTracer Tracer m (RealPoint blk, BlockNo)
selTracer Tracer m (LedgerUpdate blk)
updatesTracer Tracer m (TracePipeliningEvent blk)
pipeliningTracer
      NodeDBs (StrictTMVar m MockFS)
nodeDBs CoreNodeId
_coreNodeId =
        let args :: ChainDbArgs Identity m blk
args = MinimalChainDbArgs m blk -> ChainDbArgs Identity m blk
forall (m :: * -> *) blk.
(MonadThrow m, MonadSTM m, ConsensusProtocol (BlockProtocol blk),
 PrimMonad m) =>
MinimalChainDbArgs m blk -> Complete ChainDbArgs m blk
fromMinimalChainDbArgs MinimalChainDbArgs {
                mcdbTopLevelConfig :: TopLevelConfig blk
mcdbTopLevelConfig = TopLevelConfig blk
cfg
              , mcdbChunkInfo :: ChunkInfo
mcdbChunkInfo      = EpochSize -> ChunkInfo
ImmutableDB.simpleChunkInfo EpochSize
epochSize0
              , mcdbInitLedger :: ExtLedgerState blk ValuesMK
mcdbInitLedger     = ExtLedgerState blk ValuesMK
initLedger
              , mcdbRegistry :: ResourceRegistry m
mcdbRegistry       = ResourceRegistry m
registry
              , mcdbNodeDBs :: NodeDBs (StrictTMVar m MockFS)
mcdbNodeDBs        = NodeDBs (StrictTMVar m MockFS)
nodeDBs
              }
            tr :: Tracer m (TraceEvent blk)
tr = Tracer m (TraceEvent blk)
instrumentationTracer Tracer m (TraceEvent blk)
-> Tracer m (TraceEvent blk) -> Tracer m (TraceEvent blk)
forall a. Semigroup a => a -> a -> a
<> Tracer m (TraceEvent blk)
forall (m :: * -> *) a. (Applicative m, Show a) => Tracer m a
nullDebugTracer
        in ChainDbArgs Identity m blk
args { cdbImmDbArgs = (cdbImmDbArgs args) {
                      ImmutableDB.immCheckIntegrity = nodeCheckIntegrity (configStorage cfg)
                    , ImmutableDB.immTracer = TraceImmutableDBEvent >$< tr
                    }
                , cdbVolDbArgs = (cdbVolDbArgs args) {
                      VolatileDB.volCheckIntegrity = nodeCheckIntegrity (configStorage cfg)
                    , VolatileDB.volTracer = TraceVolatileDBEvent >$< tr
                    }
                , cdbLgrDbArgs = (cdbLgrDbArgs args) {
                      LedgerDB.lgrTracer = TraceLedgerDBEvent >$< tr
                    }
                , cdbsArgs = (cdbsArgs args) {
                     -- TODO: Vary cdbsGcDelay, cdbsGcInterval, cdbsBlockToAddSize
                      cdbsGcDelay = 0
                    , cdbsTracer = instrumentationTracer <> nullDebugTracer
                    }
                }
      where
        prj :: AnchoredFragment block -> BlockNo
prj AnchoredFragment block
af = case AnchoredFragment block -> WithOrigin BlockNo
forall block.
HasHeader block =>
AnchoredFragment block -> WithOrigin BlockNo
AF.headBlockNo AnchoredFragment block
af of
            At BlockNo
bno -> BlockNo
bno
            WithOrigin BlockNo
Origin -> String -> BlockNo
forall a. HasCallStack => String -> a
error String
"selTracer"

        -- prop_general relies on this tracer
        instrumentationTracer :: Tracer m (TraceEvent blk)
instrumentationTracer = (TraceEvent blk -> m ()) -> Tracer m (TraceEvent blk)
forall (m :: * -> *) a. (a -> m ()) -> Tracer m a
Tracer ((TraceEvent blk -> m ()) -> Tracer m (TraceEvent blk))
-> (TraceEvent blk -> m ()) -> Tracer m (TraceEvent blk)
forall a b. (a -> b) -> a -> b
$ \case
          ChainDB.TraceAddBlockEvent
              (ChainDB.AddBlockValidation (ChainDB.InvalidBlock ExtValidationError blk
e RealPoint blk
p))
              -> Tracer m (RealPoint blk, ExtValidationError blk)
-> (RealPoint blk, ExtValidationError blk) -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (RealPoint blk, ExtValidationError blk)
invalidTracer (RealPoint blk
p, ExtValidationError blk
e)

          ChainDB.TraceAddBlockEvent
              (ChainDB.AddedBlockToVolatileDB RealPoint blk
p BlockNo
bno IsEBB
IsNotEBB Enclosing' ()
FallingEdge)
              -> Tracer m (RealPoint blk, BlockNo)
-> (RealPoint blk, BlockNo) -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (RealPoint blk, BlockNo)
addTracer (RealPoint blk
p, BlockNo
bno)

          ChainDB.TraceAddBlockEvent
              (ChainDB.AddedToCurrentChain [LedgerEvent blk]
events SelectionChangedInfo blk
p AnchoredFragment (Header blk)
_old AnchoredFragment (Header blk)
new)
              -> let ([LedgerWarning blk]
warnings, [LedgerUpdate blk]
updates) = [LedgerEvent blk] -> ([LedgerWarning blk], [LedgerUpdate blk])
forall blk.
[LedgerEvent blk] -> ([LedgerWarning blk], [LedgerUpdate blk])
partitionLedgerEvents [LedgerEvent blk]
events in
                 Either String () -> m () -> m ()
forall a. HasCallStack => Either String () -> a -> a
assertWithMsg ([LedgerWarning blk] -> Either String ()
forall a. Show a => [a] -> Either String ()
noWarnings [LedgerWarning blk]
warnings) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                   (LedgerUpdate blk -> m ()) -> [LedgerUpdate blk] -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (Tracer m (LedgerUpdate blk) -> LedgerUpdate blk -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (LedgerUpdate blk)
updatesTracer) [LedgerUpdate blk]
updates
                   Tracer m (RealPoint blk, BlockNo)
-> (RealPoint blk, BlockNo) -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (RealPoint blk, BlockNo)
selTracer (SelectionChangedInfo blk -> RealPoint blk
forall blk. SelectionChangedInfo blk -> RealPoint blk
ChainDB.newTipPoint SelectionChangedInfo blk
p, AnchoredFragment (Header blk) -> BlockNo
forall {block}.
HasHeader block =>
AnchoredFragment block -> BlockNo
prj AnchoredFragment (Header blk)
new)
          ChainDB.TraceAddBlockEvent
              (ChainDB.SwitchedToAFork [LedgerEvent blk]
events SelectionChangedInfo blk
p AnchoredFragment (Header blk)
_old AnchoredFragment (Header blk)
new)
              -> let ([LedgerWarning blk]
warnings, [LedgerUpdate blk]
updates) = [LedgerEvent blk] -> ([LedgerWarning blk], [LedgerUpdate blk])
forall blk.
[LedgerEvent blk] -> ([LedgerWarning blk], [LedgerUpdate blk])
partitionLedgerEvents [LedgerEvent blk]
events in
                 Either String () -> m () -> m ()
forall a. HasCallStack => Either String () -> a -> a
assertWithMsg ([LedgerWarning blk] -> Either String ()
forall a. Show a => [a] -> Either String ()
noWarnings [LedgerWarning blk]
warnings) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                   (LedgerUpdate blk -> m ()) -> [LedgerUpdate blk] -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (Tracer m (LedgerUpdate blk) -> LedgerUpdate blk -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (LedgerUpdate blk)
updatesTracer) [LedgerUpdate blk]
updates
                   Tracer m (RealPoint blk, BlockNo)
-> (RealPoint blk, BlockNo) -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (RealPoint blk, BlockNo)
selTracer (SelectionChangedInfo blk -> RealPoint blk
forall blk. SelectionChangedInfo blk -> RealPoint blk
ChainDB.newTipPoint SelectionChangedInfo blk
p, AnchoredFragment (Header blk) -> BlockNo
forall {block}.
HasHeader block =>
AnchoredFragment block -> BlockNo
prj AnchoredFragment (Header blk)
new)

          ChainDB.TraceAddBlockEvent
              (ChainDB.PipeliningEvent TracePipeliningEvent blk
e)
              -> Tracer m (TracePipeliningEvent blk)
-> TracePipeliningEvent blk -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (TracePipeliningEvent blk)
pipeliningTracer TracePipeliningEvent blk
e

          TraceEvent blk
_   -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

    -- We don't expect any ledger warnings
    -- (that would indicate node misconfiguration in the tests)
    noWarnings :: Show a => [a] -> Either String ()
    noWarnings :: forall a. Show a => [a] -> Either String ()
noWarnings [] = () -> Either String ()
forall a b. b -> Either a b
Right ()
    noWarnings [a]
ws = String -> Either String ()
forall a b. a -> Either a b
Left (String -> Either String ()) -> String -> Either String ()
forall a b. (a -> b) -> a -> b
$ String
"Unexpected warnings: " String -> ShowS
forall a. [a] -> [a] -> [a]
++ [a] -> String
forall a. Show a => a -> String
show [a]
ws

    -- Augment a tracer message with the node which produces it.
    _decorateId :: CoreNodeId -> Tracer m String -> Tracer m String
    _decorateId :: CoreNodeId -> Tracer m String -> Tracer m String
_decorateId (CoreNodeId Word64
cid) = ShowS -> Tracer m String -> Tracer m String
forall a' a. (a' -> a) -> Tracer m a -> Tracer m a'
forall (f :: * -> *) a' a.
Contravariant f =>
(a' -> a) -> f a -> f a'
contramap (ShowS -> Tracer m String -> Tracer m String)
-> ShowS -> Tracer m String -> Tracer m String
forall a b. (a -> b) -> a -> b
$ \String
s ->
        Word64 -> String
forall a. Show a => a -> String
show Word64
cid String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" | " String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
s

    forkNode
      :: HasCallStack
      => CoreNodeId
      -> OracularClock m
      -> SlotNo
      -> ResourceRegistry m
      -> ProtocolInfo blk
      -> m [BlockForging m blk]
      -> NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
      -> [GenTx blk]
         -- ^ valid transactions the node should immediately propagate
      -> m ( NodeKernel m NodeId Void blk
           , LimitedApp m NodeId      blk
           )
    forkNode :: HasCallStack =>
CoreNodeId
-> OracularClock m
-> SlotNo
-> ResourceRegistry m
-> ProtocolInfo blk
-> m [BlockForging m blk]
-> NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
-> [GenTx blk]
-> m (NodeKernel m NodeId Void blk, LimitedApp m NodeId blk)
forkNode CoreNodeId
coreNodeId OracularClock m
clock SlotNo
joinSlot ResourceRegistry m
registry ProtocolInfo blk
pInfo m [BlockForging m blk]
blockForging NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
nodeInfo [GenTx blk]
txs0 = do
      let ProtocolInfo{TopLevelConfig blk
ExtLedgerState blk ValuesMK
pInfoInitLedger :: forall b. ProtocolInfo b -> ExtLedgerState b ValuesMK
pInfoConfig :: forall b. ProtocolInfo b -> TopLevelConfig b
pInfoConfig :: TopLevelConfig blk
pInfoInitLedger :: ExtLedgerState blk ValuesMK
..} = ProtocolInfo blk
pInfo

      let NodeInfo
            { NodeEvents blk (Tracer m)
nodeInfoEvents :: forall blk db (ev :: * -> *).
NodeInfo blk db ev -> NodeEvents blk ev
nodeInfoEvents :: NodeEvents blk (Tracer m)
nodeInfoEvents
            , NodeDBs (StrictTMVar m MockFS)
nodeInfoDBs :: NodeDBs (StrictTMVar m MockFS)
nodeInfoDBs :: forall blk db (ev :: * -> *). NodeInfo blk db ev -> NodeDBs db
nodeInfoDBs
            } = NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
nodeInfo

      -- prop_general relies on these tracers
      let invalidTracer :: Tracer m (RealPoint blk, ExtValidationError blk)
invalidTracer = NodeEvents blk (Tracer m)
-> Tracer m (RealPoint blk, ExtValidationError blk)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (RealPoint blk, ExtValidationError blk)
nodeEventsInvalids NodeEvents blk (Tracer m)
nodeInfoEvents
          updatesTracer :: Tracer m (LedgerUpdate blk)
updatesTracer = NodeEvents blk (Tracer m) -> Tracer m (LedgerUpdate blk)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (LedgerUpdate blk)
nodeEventsUpdates  NodeEvents blk (Tracer m)
nodeInfoEvents
          wrapTracer :: Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
wrapTracer Tracer m (SlotNo, RealPoint blk, BlockNo)
tr   = ((RealPoint blk, BlockNo) -> m ())
-> Tracer m (RealPoint blk, BlockNo)
forall (m :: * -> *) a. (a -> m ()) -> Tracer m a
Tracer (((RealPoint blk, BlockNo) -> m ())
 -> Tracer m (RealPoint blk, BlockNo))
-> ((RealPoint blk, BlockNo) -> m ())
-> Tracer m (RealPoint blk, BlockNo)
forall a b. (a -> b) -> a -> b
$ \(RealPoint blk
p, BlockNo
bno) -> do
            s <- OracularClock m -> m SlotNo
forall (m :: * -> *). OracularClock m -> m SlotNo
OracularClock.getCurrentSlot OracularClock m
clock
            traceWith tr (s, p, bno)
          addTracer :: Tracer m (RealPoint blk, BlockNo)
addTracer        = Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
wrapTracer (Tracer m (SlotNo, RealPoint blk, BlockNo)
 -> Tracer m (RealPoint blk, BlockNo))
-> Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
forall a b. (a -> b) -> a -> b
$ NodeEvents blk (Tracer m)
-> Tracer m (SlotNo, RealPoint blk, BlockNo)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, RealPoint blk, BlockNo)
nodeEventsAdds NodeEvents blk (Tracer m)
nodeInfoEvents
          selTracer :: Tracer m (RealPoint blk, BlockNo)
selTracer        = Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
wrapTracer (Tracer m (SlotNo, RealPoint blk, BlockNo)
 -> Tracer m (RealPoint blk, BlockNo))
-> Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
forall a b. (a -> b) -> a -> b
$ NodeEvents blk (Tracer m)
-> Tracer m (SlotNo, RealPoint blk, BlockNo)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, RealPoint blk, BlockNo)
nodeEventsSelects NodeEvents blk (Tracer m)
nodeInfoEvents
          headerAddTracer :: Tracer m (RealPoint blk, BlockNo)
headerAddTracer  = Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
wrapTracer (Tracer m (SlotNo, RealPoint blk, BlockNo)
 -> Tracer m (RealPoint blk, BlockNo))
-> Tracer m (SlotNo, RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
forall a b. (a -> b) -> a -> b
$ NodeEvents blk (Tracer m)
-> Tracer m (SlotNo, RealPoint blk, BlockNo)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, RealPoint blk, BlockNo)
nodeEventsHeaderAdds NodeEvents blk (Tracer m)
nodeInfoEvents
          pipeliningTracer :: Tracer m (TracePipeliningEvent blk)
pipeliningTracer = NodeEvents blk (Tracer m) -> Tracer m (TracePipeliningEvent blk)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (TracePipeliningEvent blk)
nodeEventsPipelining NodeEvents blk (Tracer m)
nodeInfoEvents
      let chainDbArgs :: ChainDbArgs Identity m blk
chainDbArgs = ResourceRegistry m
-> TopLevelConfig blk
-> ExtLedgerState blk ValuesMK
-> Tracer m (RealPoint blk, ExtValidationError blk)
-> Tracer m (RealPoint blk, BlockNo)
-> Tracer m (RealPoint blk, BlockNo)
-> Tracer m (LedgerUpdate blk)
-> Tracer m (TracePipeliningEvent blk)
-> NodeDBs (StrictTMVar m MockFS)
-> CoreNodeId
-> ChainDbArgs Identity m blk
mkArgs
            ResourceRegistry m
registry
            TopLevelConfig blk
pInfoConfig ExtLedgerState blk ValuesMK
pInfoInitLedger
            Tracer m (RealPoint blk, ExtValidationError blk)
invalidTracer
            Tracer m (RealPoint blk, BlockNo)
addTracer
            Tracer m (RealPoint blk, BlockNo)
selTracer
            Tracer m (LedgerUpdate blk)
updatesTracer
            Tracer m (TracePipeliningEvent blk)
pipeliningTracer
            NodeDBs (StrictTMVar m MockFS)
nodeInfoDBs
            CoreNodeId
coreNodeId
      chainDB <- (ResourceKey m, ChainDB m blk) -> ChainDB m blk
forall a b. (a, b) -> b
snd ((ResourceKey m, ChainDB m blk) -> ChainDB m blk)
-> m (ResourceKey m, ChainDB m blk) -> m (ChainDB m blk)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>
        ResourceRegistry m
-> (ResourceId -> m (ChainDB m blk))
-> (ChainDB m blk -> m ())
-> m (ResourceKey m, ChainDB m blk)
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 (m (ChainDB m blk) -> ResourceId -> m (ChainDB m blk)
forall a b. a -> b -> a
const (ChainDbArgs Identity m blk -> m (ChainDB m blk)
forall (m :: * -> *) blk.
(IOLike m, LedgerSupportsProtocol blk,
 BlockSupportsDiffusionPipelining blk, InspectLedger blk,
 HasHardForkHistory blk, ConvertRawHash blk,
 SerialiseDiskConstraints blk, LedgerSupportsLedgerDB blk) =>
Complete ChainDbArgs m blk -> m (ChainDB m blk)
ChainDB.openDB ChainDbArgs Identity m blk
chainDbArgs)) ChainDB m blk -> m ()
forall (m :: * -> *) blk. ChainDB m blk -> m ()
ChainDB.closeDB

      let customForgeBlock ::
               BlockForging m blk
            -> TopLevelConfig blk
            -> BlockNo
            -> SlotNo
            -> TickedLedgerState blk mk
            -> [Validated (GenTx blk)]
            -> IsLeader (BlockProtocol blk)
            -> m blk
          customForgeBlock BlockForging m blk
origBlockForging TopLevelConfig blk
cfg' BlockNo
currentBno SlotNo
currentSlot TickedLedgerState blk mk
tickedLdgSt [Validated (GenTx blk)]
txs IsLeader (BlockProtocol blk)
prf = do
            let currentEpoch :: EpochNo
currentEpoch = Future -> SlotNo -> EpochNo
HFF.futureSlotToEpoch Future
future SlotNo
currentSlot

            -- EBBs are only ever possible in the first era
            let inFirstEra :: Bool
inFirstEra = Future -> EpochNo -> Bool
HFF.futureEpochInFirstEra Future
future EpochNo
currentEpoch

            let ebbSlot :: SlotNo
                ebbSlot :: SlotNo
ebbSlot = Word64 -> SlotNo
SlotNo (Word64 -> SlotNo) -> Word64 -> SlotNo
forall a b. (a -> b) -> a -> b
$ Word64
x Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
* Word64
y
                  where
                    EpochNo   Word64
x = EpochNo
currentEpoch
                    EpochSize Word64
y = EpochSize
epochSize0

            let p :: Point blk
                p :: Point blk
p = Point (Ticked (LedgerState blk)) -> Point blk
forall {k1} {k2} (b :: k1) (b' :: k2).
Coercible (HeaderHash b) (HeaderHash b') =>
Point b -> Point b'
castPoint (Point (Ticked (LedgerState blk)) -> Point blk)
-> Point (Ticked (LedgerState blk)) -> Point blk
forall a b. (a -> b) -> a -> b
$ TickedLedgerState blk mk -> Point (Ticked (LedgerState blk))
forall (mk :: MapKind).
Ticked (LedgerState blk) mk -> Point (Ticked (LedgerState blk))
forall (l :: LedgerStateKind) (mk :: MapKind).
GetTip l =>
l mk -> Point l
getTip TickedLedgerState blk mk
tickedLdgSt

            let needEBB :: Bool
needEBB = Bool
inFirstEra Bool -> Bool -> Bool
&& SlotNo -> WithOrigin SlotNo
forall t. t -> WithOrigin t
NotOrigin SlotNo
ebbSlot WithOrigin SlotNo -> WithOrigin SlotNo -> Bool
forall a. Ord a => a -> a -> Bool
> Point blk -> WithOrigin SlotNo
forall {k} (block :: k). Point block -> WithOrigin SlotNo
pointSlot Point blk
p
            case Maybe (ForgeEbbEnv blk)
mbForgeEbbEnv Maybe (ForgeEbbEnv blk) -> Maybe () -> Maybe (ForgeEbbEnv blk)
forall a b. Maybe a -> Maybe b -> Maybe a
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f a
<* Bool -> Maybe ()
forall (f :: * -> *). Alternative f => Bool -> f ()
guard Bool
needEBB of
              Maybe (ForgeEbbEnv blk)
Nothing ->
                 -- no EBB needed, forge without making one
                 BlockForging m blk
-> TopLevelConfig blk
-> BlockNo
-> SlotNo
-> TickedLedgerState blk EmptyMK
-> [Validated (GenTx blk)]
-> IsLeader (BlockProtocol blk)
-> m blk
forall (m :: * -> *) blk.
BlockForging m blk
-> TopLevelConfig blk
-> BlockNo
-> SlotNo
-> TickedLedgerState blk EmptyMK
-> [Validated (GenTx blk)]
-> IsLeader (BlockProtocol blk)
-> m blk
forgeBlock
                   BlockForging m blk
origBlockForging
                   TopLevelConfig blk
cfg'
                   BlockNo
currentBno
                   SlotNo
currentSlot
                   (TickedLedgerState blk mk -> TickedLedgerState blk EmptyMK
forall (l :: LedgerStateKind) (mk :: MapKind).
HasLedgerTables l =>
l mk -> l EmptyMK
forgetLedgerTables TickedLedgerState blk mk
tickedLdgSt)
                   [Validated (GenTx blk)]
txs
                   IsLeader (BlockProtocol blk)
prf
              Just ForgeEbbEnv blk
forgeEbbEnv -> do
                  -- The EBB shares its BlockNo with its predecessor (if
                  -- there is one)
                  let ebbBno :: BlockNo
ebbBno = case BlockNo
currentBno of
                        -- We assume this invariant:
                        --
                        -- If forging of EBBs is enabled then the node
                        -- initialization is responsible for producing any
                        -- proper non-EBB blocks with block number 0.
                        --
                        -- So this case is unreachable.
                        BlockNo
0 -> String -> BlockNo
forall a. HasCallStack => String -> a
error String
"Error, only node initialization can forge non-EBB with block number 0."
                        BlockNo
n -> BlockNo -> BlockNo
forall a. Enum a => a -> a
pred BlockNo
n
                  let ebb :: blk
ebb = ForgeEbbEnv blk
-> TopLevelConfig blk -> SlotNo -> BlockNo -> ChainHash blk -> blk
forall blk.
ForgeEbbEnv blk
-> TopLevelConfig blk -> SlotNo -> BlockNo -> ChainHash blk -> blk
forgeEBB ForgeEbbEnv blk
forgeEbbEnv TopLevelConfig blk
pInfoConfig
                              SlotNo
ebbSlot BlockNo
ebbBno (Point blk -> ChainHash blk
forall {k} (block :: k). Point block -> ChainHash block
pointHash Point blk
p)

                  -- fail if the EBB is invalid
                  -- if it is valid, we retick to the /same/ slot
                  let apply :: blk
-> Ticked (LedgerState blk) ValuesMK
-> Except (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
apply  = ComputeLedgerEvents
-> LedgerCfg (LedgerState blk)
-> blk
-> Ticked (LedgerState blk) ValuesMK
-> Except (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
forall (l :: LedgerStateKind) blk.
(ApplyBlock l blk, HasCallStack) =>
ComputeLedgerEvents
-> LedgerCfg l
-> blk
-> Ticked l ValuesMK
-> Except (LedgerErr l) (l DiffMK)
applyLedgerBlock ComputeLedgerEvents
OmitLedgerEvents (TopLevelConfig blk -> LedgerCfg (LedgerState blk)
forall blk. TopLevelConfig blk -> LedgerConfig blk
configLedger TopLevelConfig blk
pInfoConfig)
                      tables :: LedgerTables (Ticked (LedgerState blk)) ValuesMK
tables = LedgerTables (Ticked (LedgerState blk)) ValuesMK
forall (mk :: MapKind) (l :: LedgerStateKind).
(ZeroableMK mk, LedgerTableConstraints l) =>
LedgerTables l mk
emptyLedgerTables   -- EBBs need no input tables
                  tickedLdgSt' <- case Except (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
-> Either (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
forall e a. Except e a -> Either e a
Exc.runExcept (Except (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
 -> Either (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK))
-> Except (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
-> Either (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
forall a b. (a -> b) -> a -> b
$ blk
-> Ticked (LedgerState blk) ValuesMK
-> Except (LedgerErr (LedgerState blk)) (LedgerState blk DiffMK)
apply blk
ebb (TickedLedgerState blk mk
tickedLdgSt TickedLedgerState blk mk
-> LedgerTables (Ticked (LedgerState blk)) ValuesMK
-> Ticked (LedgerState blk) ValuesMK
forall (mk :: MapKind) (any :: MapKind).
(CanMapMK mk, CanMapKeysMK mk, ZeroableMK mk) =>
Ticked (LedgerState blk) any
-> LedgerTables (Ticked (LedgerState blk)) mk
-> Ticked (LedgerState blk) mk
forall (l :: LedgerStateKind) (mk :: MapKind) (any :: MapKind).
(HasLedgerTables l, CanMapMK mk, CanMapKeysMK mk, ZeroableMK mk) =>
l any -> LedgerTables l mk -> l mk
`withLedgerTables` LedgerTables (Ticked (LedgerState blk)) ValuesMK
tables) of
                    Left LedgerErr (LedgerState blk)
e   -> JitEbbError blk -> m (Ticked (LedgerState blk) DiffMK)
forall a e. (HasCallStack, Exception e) => e -> a
Exn.throw (JitEbbError blk -> m (Ticked (LedgerState blk) DiffMK))
-> JitEbbError blk -> m (Ticked (LedgerState blk) DiffMK)
forall a b. (a -> b) -> a -> b
$ forall blk. LedgerError blk -> JitEbbError blk
JitEbbError @blk LedgerErr (LedgerState blk)
e
                    Right LedgerState blk DiffMK
st -> Ticked (LedgerState blk) DiffMK
-> m (Ticked (LedgerState blk) DiffMK)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Ticked (LedgerState blk) DiffMK
 -> m (Ticked (LedgerState blk) DiffMK))
-> Ticked (LedgerState blk) DiffMK
-> m (Ticked (LedgerState blk) DiffMK)
forall a b. (a -> b) -> a -> b
$ ComputeLedgerEvents
-> LedgerCfg (LedgerState blk)
-> SlotNo
-> LedgerState blk EmptyMK
-> Ticked (LedgerState blk) DiffMK
forall (l :: LedgerStateKind).
IsLedger l =>
ComputeLedgerEvents
-> LedgerCfg l -> SlotNo -> l EmptyMK -> Ticked l DiffMK
applyChainTick ComputeLedgerEvents
OmitLedgerEvents
                                        (TopLevelConfig blk -> LedgerCfg (LedgerState blk)
forall blk. TopLevelConfig blk -> LedgerConfig blk
configLedger TopLevelConfig blk
pInfoConfig)
                                        SlotNo
currentSlot
                                        (LedgerState blk DiffMK -> LedgerState blk EmptyMK
forall (l :: LedgerStateKind) (mk :: MapKind).
HasLedgerTables l =>
l mk -> l EmptyMK
forgetLedgerTables LedgerState blk DiffMK
st)

                  -- forge the block usings the ledger state that includes
                  -- the EBB
                  blk <- forgeBlock
                           origBlockForging
                           cfg'
                           currentBno
                           currentSlot
                           (forgetLedgerTables tickedLdgSt')
                           txs
                           prf

                  -- If the EBB or the subsequent block is invalid, then the
                  -- ChainDB will reject it as invalid, and
                  -- 'Test.ThreadNet.General.prop_general' will eventually fail
                  -- because of a block rejection.
                  void $ ChainDB.addBlock chainDB InvalidBlockPunishment.noPunishment ebb
                  pure blk

      -- This variable holds the number of the earliest slot in which the
      -- crucial txs have not yet been added. In other words, it holds the
      -- successor of the number of the latest slot in which the crucial txs
      -- have been added.
      --
      -- Key facts: The thread that adds the crucial transactions updates this
      -- variable, and the forge tracer for 'TraceNodeIsLeader' blocks on it.
      (unblockForge, blockOnCrucial) <- do
        var <- uncheckedNewTVarM 0
        pure
          ( \SlotNo
s -> do
              StrictTVar m SlotNo -> (SlotNo -> SlotNo) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m SlotNo
var (SlotNo -> SlotNo
forall a. Enum a => a -> a
succ SlotNo
s SlotNo -> SlotNo -> SlotNo
forall a. Ord a => a -> a -> a
`max`)
          , \SlotNo
s -> do
              sentinel <- StrictTVar m SlotNo -> STM m SlotNo
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m SlotNo
var
              check $ s < sentinel
          )

      let -- prop_general relies on these tracers
          instrumentationTracers = Tracers m (ConnectionId NodeId) Any blk
forall (m :: * -> *) remotePeer localPeer blk.
Monad m =>
Tracers m remotePeer localPeer blk
nullTracers
                { chainSyncClientTracer = Tracer $ \case
                    TraceLabelPeer ConnectionId NodeId
_ (CSClient.TraceDownloadedHeader Header blk
hdr)
                      -> case Header blk -> Point (Header blk)
forall block. HasHeader block => block -> Point block
blockPoint Header blk
hdr of
                            Point (Header blk)
GenesisPoint   -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                            BlockPoint SlotNo
s HeaderHash (Header blk)
h ->
                                -- TODO include tip in TraceDownloadedHeader
                                -- and only trace if hdr == tip?
                                Tracer m (RealPoint blk, BlockNo)
-> (RealPoint blk, BlockNo) -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (RealPoint blk, BlockNo)
headerAddTracer
                                  (SlotNo -> HeaderHash blk -> RealPoint blk
forall blk. SlotNo -> HeaderHash blk -> RealPoint blk
RealPoint SlotNo
s HeaderHash blk
HeaderHash (Header blk)
h, Header blk -> BlockNo
forall b. HasHeader b => b -> BlockNo
blockNo Header blk
hdr)
                    TraceLabelPeer
  (ConnectionId NodeId) (TraceChainSyncClientEvent blk)
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                , forgeTracer           = Tracer $ \(TraceLabelCreds Text
_ TraceForgeEvent blk
ev) -> do
                    Tracer m (TraceForgeEvent blk) -> TraceForgeEvent blk -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith (NodeEvents blk (Tracer m) -> Tracer m (TraceForgeEvent blk)
forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (TraceForgeEvent blk)
nodeEventsForges NodeEvents blk (Tracer m)
nodeInfoEvents) TraceForgeEvent blk
ev
                    case TraceForgeEvent blk
ev of
                      TraceNodeIsLeader SlotNo
s -> 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
$ SlotNo -> STM m ()
blockOnCrucial SlotNo
s
                      TraceForgeEvent blk
_                   -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                }

          -- traces the node's local events other than those from the -- ChainDB
          tracers = Tracers' (ConnectionId NodeId) Void blk (Tracer m)
instrumentationTracers Tracers' (ConnectionId NodeId) Void blk (Tracer m)
-> Tracers' (ConnectionId NodeId) Void blk (Tracer m)
-> Tracers' (ConnectionId NodeId) Void blk (Tracer m)
forall a. Semigroup a => a -> a -> a
<> Tracers' (ConnectionId NodeId) Void blk (Tracer m)
forall (m :: * -> *) peer blk.
(Monad m, Show peer, LedgerSupportsProtocol blk,
 TracingConstraints blk) =>
Tracers m peer Void blk
nullDebugTracers

      let -- use a backoff delay of exactly one slot length (which the
          -- 'OracularClock' always knows) for the following reasons
          --
          -- o It gives the node a chance to sync some blocks so that it will
          --   eventually not need to backoff
          --
          -- o It maintains the invariant that the node's activities all happen "
          --   during " a slot onset
          --
          -- o It avoids causing the node to miss a slot it could have
          --   nominally lead. EG If we used a backoff of two slot durations,
          --   then it might have synced during the first slot and then been
          --   able to productively lead the second slot had it not still been
          --   asleep.
          --
          -- o We assume a node will only backoff when it joins late and only
          --   until it syncs enough of the net's existing common prefix.
          hfbtBackoffDelay =
              NominalDiffTime -> BackoffDelay
BackoffDelay (NominalDiffTime -> BackoffDelay)
-> m NominalDiffTime -> m BackoffDelay
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> OracularClock m -> m NominalDiffTime
forall (m :: * -> *). OracularClock m -> m NominalDiffTime
OracularClock.delayUntilNextSlot OracularClock m
clock
      btime <- hardForkBlockchainTime HardForkBlockchainTimeArgs
        { hfbtBackoffDelay
        , hfbtGetLedgerState =
            ledgerState <$> ChainDB.getCurrentLedger chainDB
        , hfbtLedgerConfig   = configLedger pInfoConfig
        , hfbtRegistry       = registry
        , hfbtSystemTime     = OracularClock.finiteSystemTime clock
        , hfbtTracer         =
            contramap
              -- We don't really have a SystemStart in the tests
              (fmap (fromRelativeTime (SystemStart dawnOfTime)))
              (blockchainTimeTracer tracers)
        , hfbtMaxClockRewind = secondsToNominalDiffTime 0
        }

      let rng = case Seed
seed of
                    Seed Int
s -> Int -> StdGen
mkStdGen Int
s
          (kaRng, psRng) = split rng
      publicPeerSelectionStateVar <- makePublicPeerSelectionStateVar
      let nodeKernelArgs = NodeKernelArgs
            { Tracers' (ConnectionId NodeId) Void blk (Tracer m)
tracers :: Tracers' (ConnectionId NodeId) Void blk (Tracer m)
tracers :: Tracers' (ConnectionId NodeId) Void blk (Tracer m)
tracers
            , ResourceRegistry m
registry :: ResourceRegistry m
registry :: ResourceRegistry m
registry
            , cfg :: TopLevelConfig blk
cfg                     = TopLevelConfig blk
pInfoConfig
            , BlockchainTime m
btime :: BlockchainTime m
btime :: BlockchainTime m
btime
            , ChainDB m blk
chainDB :: ChainDB m blk
chainDB :: ChainDB m blk
chainDB
            , initChainDB :: StorageConfig blk -> InitChainDB m blk -> m ()
initChainDB             = StorageConfig blk -> InitChainDB m blk -> m ()
forall blk (m :: * -> *).
(NodeInitStorage blk, IOLike m) =>
StorageConfig blk -> InitChainDB m blk -> m ()
forall (m :: * -> *).
IOLike m =>
StorageConfig blk -> InitChainDB m blk -> m ()
nodeInitChainDB
            , chainSyncFutureCheck :: SomeHeaderInFutureCheck m blk
chainSyncFutureCheck    =
                  ClockSkew -> SystemTime m -> SomeHeaderInFutureCheck m blk
forall blk (m :: * -> *).
(HasHeader blk, HasHeader (Header blk), HasHardForkHistory blk,
 MonadDelay m) =>
ClockSkew -> SystemTime m -> SomeHeaderInFutureCheck m blk
InFutureCheck.realHeaderInFutureCheck
                    ClockSkew
InFutureCheck.defaultClockSkew
                    (OracularClock m -> SystemTime m
forall (m :: * -> *). OracularClock m -> SystemTime m
OracularClock.finiteSystemTime OracularClock m
clock)
            , chainSyncHistoricityCheck :: m GsmState -> HistoricityCheck m blk
chainSyncHistoricityCheck = \m GsmState
_getGsmState -> HistoricityCheck m blk
forall (m :: * -> *) blk. Applicative m => HistoricityCheck m blk
HistoricityCheck.noCheck
            , blockFetchSize :: Header blk -> SizeInBytes
blockFetchSize          = Header blk -> SizeInBytes
forall blk.
SerialiseNodeToNodeConstraints blk =>
Header blk -> SizeInBytes
estimateBlockSize
            , mempoolCapacityOverride :: MempoolCapacityBytesOverride
mempoolCapacityOverride = MempoolCapacityBytesOverride
NoMempoolCapacityBytesOverride
            , keepAliveRng :: StdGen
keepAliveRng            = StdGen
kaRng
            , peerSharingRng :: StdGen
peerSharingRng          = StdGen
psRng
            , miniProtocolParameters :: MiniProtocolParameters
miniProtocolParameters  = MiniProtocolParameters {
                  chainSyncPipeliningHighMark :: Word16
chainSyncPipeliningHighMark = Word16
4,
                  chainSyncPipeliningLowMark :: Word16
chainSyncPipeliningLowMark  = Word16
2,
                  blockFetchPipeliningMax :: Word16
blockFetchPipeliningMax     = Word16
10,
                  txSubmissionMaxUnacked :: NumTxIdsToAck
txSubmissionMaxUnacked      = NumTxIdsToAck
1000 -- TODO ?
                }
            , blockFetchConfiguration :: BlockFetchConfiguration
blockFetchConfiguration = BlockFetchConfiguration {
                  bfcMaxConcurrencyBulkSync :: Word
bfcMaxConcurrencyBulkSync = Word
1
                , bfcMaxConcurrencyDeadline :: Word
bfcMaxConcurrencyDeadline = Word
2
                , bfcMaxRequestsInflight :: Word
bfcMaxRequestsInflight    = Word
10
                , bfcDecisionLoopIntervalPraos :: DiffTime
bfcDecisionLoopIntervalPraos = DiffTime
0.0 -- Mock testsuite can use sub-second slot
                , bfcDecisionLoopIntervalGenesis :: DiffTime
bfcDecisionLoopIntervalGenesis = DiffTime
0.0 -- interval which doesn't play nice with
                                                        -- blockfetch descision interval.
                , bfcSalt :: Int
bfcSalt                   = Int
0
                , bfcGenesisBFConfig :: GenesisBlockFetchConfiguration
bfcGenesisBFConfig        = GenesisConfig -> GenesisBlockFetchConfiguration
gcBlockFetchConfig GenesisConfig
enableGenesisConfigDefault
                }
            , gsmArgs :: GsmNodeKernelArgs m blk
gsmArgs                 = GSM.GsmNodeKernelArgs {
                  gsmAntiThunderingHerd :: StdGen
gsmAntiThunderingHerd  = StdGen
kaRng
                , gsmDurationUntilTooOld :: Maybe (WrapDurationUntilTooOld m blk)
gsmDurationUntilTooOld = Maybe (WrapDurationUntilTooOld m blk)
forall a. Maybe a
Nothing
                , gsmMarkerFileView :: MarkerFileView m
gsmMarkerFileView      = GSM.MarkerFileView {
                      touchMarkerFile :: m ()
touchMarkerFile  = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                    , removeMarkerFile :: m ()
removeMarkerFile = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                    , hasMarkerFile :: m Bool
hasMarkerFile    = Bool -> m Bool
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
                    }
                , gsmMinCaughtUpDuration :: NominalDiffTime
gsmMinCaughtUpDuration = NominalDiffTime
0
                }
            , getUseBootstrapPeers :: STM m UseBootstrapPeers
getUseBootstrapPeers = UseBootstrapPeers -> STM m UseBootstrapPeers
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure UseBootstrapPeers
DontUseBootstrapPeers
            , StrictTVar m (PublicPeerSelectionState NodeId)
publicPeerSelectionStateVar :: StrictTVar m (PublicPeerSelectionState NodeId)
publicPeerSelectionStateVar :: StrictTVar m (PublicPeerSelectionState NodeId)
publicPeerSelectionStateVar
            , genesisArgs :: GenesisNodeKernelArgs m blk
genesisArgs          = GenesisNodeKernelArgs {
                  gnkaLoEAndGDDArgs :: LoEAndGDDConfig (LoEAndGDDNodeKernelArgs m blk)
gnkaLoEAndGDDArgs = LoEAndGDDConfig (LoEAndGDDNodeKernelArgs m blk)
forall a. LoEAndGDDConfig a
LoEAndGDDDisabled
                }
            , getDiffusionPipeliningSupport :: DiffusionPipeliningSupport
getDiffusionPipeliningSupport = DiffusionPipeliningSupport
DiffusionPipeliningOn
            }

      nodeKernel <- initNodeKernel nodeKernelArgs

      blockForging' <-
            map (\BlockForging m blk
bf -> BlockForging m blk
bf { forgeBlock = customForgeBlock bf })
        <$> blockForging
      setBlockForging nodeKernel blockForging'

      let mempool = NodeKernel m NodeId Void blk -> Mempool m blk
forall (m :: * -> *) addrNTN addrNTC blk.
NodeKernel m addrNTN addrNTC blk -> Mempool m blk
getMempool NodeKernel m NodeId Void blk
nodeKernel
      let app = NodeKernel m NodeId Void blk
-> Tracers m NodeId blk CodecError
-> (NodeToNodeVersion
    -> Codecs
         blk
         NodeId
         CodecError
         m
         ByteString
         ByteString
         ByteString
         ByteString
         (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
         (AnyMessage KeepAlive)
         (AnyMessage (PeerSharing NodeId)))
-> ByteLimits
     ByteString
     ByteString
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
     (AnyMessage KeepAlive)
-> m ChainSyncTimeout
-> ChainSyncLoPBucketConfig
-> CSJConfig
-> ReportPeerMetrics m (ConnectionId NodeId)
-> Handlers m NodeId blk
-> Apps
     m
     NodeId
     ByteString
     ByteString
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
     (AnyMessage KeepAlive)
     (AnyMessage (PeerSharing NodeId))
     NodeToNodeInitiatorResult
     ()
forall (m :: * -> *) addrNTN addrNTC blk e bCS bBF bTX bKA bPS.
(IOLike m, MonadTimer m, Ord addrNTN, Exception e,
 LedgerSupportsProtocol blk, ShowProxy blk, ShowProxy (Header blk),
 ShowProxy (TxId (GenTx blk)), ShowProxy (GenTx blk)) =>
NodeKernel m addrNTN addrNTC blk
-> Tracers m addrNTN blk e
-> (NodeToNodeVersion
    -> Codecs blk addrNTN e m bCS bCS bBF bBF bTX bKA bPS)
-> ByteLimits bCS bBF bTX bKA
-> m ChainSyncTimeout
-> ChainSyncLoPBucketConfig
-> CSJConfig
-> ReportPeerMetrics m (ConnectionId addrNTN)
-> Handlers m addrNTN blk
-> Apps m addrNTN bCS bBF bTX bKA bPS NodeToNodeInitiatorResult ()
NTN.mkApps
                  NodeKernel m NodeId Void blk
nodeKernel
                  -- these tracers report every message sent/received by this
                  -- node
                  Tracers m NodeId blk CodecError
forall (m :: * -> *) blk peer failure.
(Monad m, HasHeader blk, TracingConstraints blk, Show peer) =>
Tracers m peer blk failure
nullDebugProtocolTracers
                  (TopLevelConfig blk
-> NodeToNodeVersion
-> Codecs
     blk
     NodeId
     CodecError
     m
     ByteString
     ByteString
     ByteString
     ByteString
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
     (AnyMessage KeepAlive)
     (AnyMessage (PeerSharing NodeId))
customNodeToNodeCodecs TopLevelConfig blk
pInfoConfig)
                  ByteLimits
  ByteString
  ByteString
  (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
  (AnyMessage KeepAlive)
forall bCS bBF bTX bKA. ByteLimits bCS bBF bTX bKA
NTN.noByteLimits
                  -- see #1882, tests that can't cope with timeouts.
                  (ChainSyncTimeout -> m ChainSyncTimeout
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ChainSyncTimeout -> m ChainSyncTimeout)
-> ChainSyncTimeout -> m ChainSyncTimeout
forall a b. (a -> b) -> a -> b
$ NTN.ChainSyncTimeout
                     { canAwaitTimeout :: Maybe DiffTime
canAwaitTimeout  = Maybe DiffTime
waitForever
                     , intersectTimeout :: Maybe DiffTime
intersectTimeout = Maybe DiffTime
waitForever
                     , mustReplyTimeout :: Maybe DiffTime
mustReplyTimeout = Maybe DiffTime
waitForever
                     , idleTimeout :: Maybe DiffTime
idleTimeout      = Maybe DiffTime
waitForever
                     })
                  ChainSyncLoPBucketConfig
CSClient.ChainSyncLoPBucketDisabled
                  CSJConfig
CSClient.CSJDisabled
                  ReportPeerMetrics m (ConnectionId NodeId)
forall (m :: * -> *) p. MonadSTM m => ReportPeerMetrics m p
nullMetric
                  -- The purpose of this test is not testing protocols, so
                  -- returning constant empty list is fine if we have thorough
                  -- tests about the peer sharing protocol itself.
                  (NodeKernelArgs m NodeId Void blk
-> NodeKernel m NodeId Void blk -> Handlers m NodeId blk
forall (m :: * -> *) blk addrNTN addrNTC.
(IOLike m, MonadTime m, MonadTimer m, LedgerSupportsMempool blk,
 HasTxId (GenTx blk), LedgerSupportsProtocol blk, Ord addrNTN,
 Hashable addrNTN) =>
NodeKernelArgs m addrNTN addrNTC blk
-> NodeKernel m addrNTN addrNTC blk -> Handlers m addrNTN blk
NTN.mkHandlers NodeKernelArgs m NodeId Void blk
nodeKernelArgs NodeKernel m NodeId Void blk
nodeKernel)

      -- Create a 'ReadOnlyForker' to be used in 'forkTxProducer'. This function
      -- needs the read-only forker to elaborate a complete UTxO set to generate
      -- transactions.
      let getForker ResourceRegistry m
rr = do
            ChainDB m blk
-> ResourceRegistry m
-> Target (Point blk)
-> m (Either GetForkerError (ReadOnlyForker' m blk))
forall (m :: * -> *) blk.
ChainDB m blk
-> ResourceRegistry m
-> Target (Point blk)
-> m (Either GetForkerError (ReadOnlyForker' m blk))
ChainDB.getReadOnlyForkerAtPoint ChainDB m blk
chainDB ResourceRegistry m
rr Target (Point blk)
forall point. Target point
VolatileTip m (Either GetForkerError (ReadOnlyForker' m blk))
-> (Either GetForkerError (ReadOnlyForker' m blk)
    -> m (ReadOnlyForker' m blk))
-> m (ReadOnlyForker' m blk)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
              Left GetForkerError
e  -> String -> m (ReadOnlyForker' m blk)
forall a. HasCallStack => String -> a
error (String -> m (ReadOnlyForker' m blk))
-> String -> m (ReadOnlyForker' m blk)
forall a b. (a -> b) -> a -> b
$ GetForkerError -> String
forall a. Show a => a -> String
show GetForkerError
e
              Right ReadOnlyForker' m blk
l -> ReadOnlyForker' m blk -> m (ReadOnlyForker' m blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ReadOnlyForker' m blk
l

      -- In practice, a robust wallet/user can persistently add a transaction
      -- until it appears on the chain. This thread adds robustness for the
      -- @txs0@ argument, which in practice contains delegation certificates
      -- that the node operator would very insistently add.
      --
      -- It's necessary here because under some circumstances a transaction in
      -- the mempool can be \"lost\" due to no fault of its own. If a dlg cert
      -- is lost, a node that rekeyed can never lead again. Moreover,
      -- promptness of certain transactions simplifies the definition of
      -- corresponding test properties: it's easier to predict whether a
      -- proposal will expire if we're ensured all votes are as prompt as
      -- possible. Lastly, the \"wallet\" might simply need to wait until
      -- enough of the chain is synced that the transaction is valid.
      --
      -- TODO Is there a risk that this will block because the 'forkTxProducer'
      -- fills up the mempool too quickly?
      forkCrucialTxs
        clock
        joinSlot
        registry
        unblockForge
        (ledgerTipPoint . ledgerState <$> ChainDB.getCurrentLedger chainDB)
        getForker
        mempool
        txs0

      forkTxProducer
        coreNodeId
        registry
        clock
        pInfoConfig
        -- Combine with the CoreNodeId, otherwise each node would generate the
        -- same transactions.
        (seed `combineWith` unCoreNodeId coreNodeId)
        -- Uses the same varRNG as the block producer, but we split the RNG
        -- each time, so this is fine.
        getForker
        mempool

      return (nodeKernel, LimitedApp app)

    customNodeToNodeCodecs
      :: TopLevelConfig blk
      -> NodeToNodeVersion
      -> NTN.Codecs blk NodeId CodecError m
           Lazy.ByteString
           Lazy.ByteString
           Lazy.ByteString
           Lazy.ByteString
           (AnyMessage (TxSubmission2 (GenTxId blk) (GenTx blk)))
           (AnyMessage KeepAlive)
           (AnyMessage (PeerSharing NodeId))
    customNodeToNodeCodecs :: TopLevelConfig blk
-> NodeToNodeVersion
-> Codecs
     blk
     NodeId
     CodecError
     m
     ByteString
     ByteString
     ByteString
     ByteString
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
     (AnyMessage KeepAlive)
     (AnyMessage (PeerSharing NodeId))
customNodeToNodeCodecs TopLevelConfig blk
cfg NodeToNodeVersion
ntnVersion = NTN.Codecs
        { cChainSyncCodec :: Codec
  (ChainSync (Header blk) (Point blk) (Tip blk))
  CodecError
  m
  ByteString
cChainSyncCodec =
            (DeserialiseFailure -> CodecError)
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     CodecError
     m
     ByteString
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec (String -> DeserialiseFailure -> CodecError
CodecBytesFailure String
"ChainSync") (Codec
   (ChainSync (Header blk) (Point blk) (Tip blk))
   DeserialiseFailure
   m
   ByteString
 -> Codec
      (ChainSync (Header blk) (Point blk) (Tip blk))
      CodecError
      m
      ByteString)
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     CodecError
     m
     ByteString
forall a b. (a -> b) -> a -> b
$
              Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec (ChainSync (Header blk) (Point blk) (Tip blk)) e m bCS
NTN.cChainSyncCodec Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
binaryProtocolCodecs
        , cChainSyncCodecSerialised :: Codec
  (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
  CodecError
  m
  ByteString
cChainSyncCodecSerialised =
            (DeserialiseFailure -> CodecError)
-> Codec
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
-> Codec
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
     CodecError
     m
     ByteString
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec (String -> DeserialiseFailure -> CodecError
CodecBytesFailure String
"ChainSyncSerialised") (Codec
   (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
   DeserialiseFailure
   m
   ByteString
 -> Codec
      (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
      CodecError
      m
      ByteString)
-> Codec
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
-> Codec
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
     CodecError
     m
     ByteString
forall a b. (a -> b) -> a -> b
$
              Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
-> Codec
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk)) e m bSCS
NTN.cChainSyncCodecSerialised Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
binaryProtocolCodecs
        , cBlockFetchCodec :: Codec (BlockFetch blk (Point blk)) CodecError m ByteString
cBlockFetchCodec =
            (DeserialiseFailure -> CodecError)
-> Codec
     (BlockFetch blk (Point blk)) DeserialiseFailure m ByteString
-> Codec (BlockFetch blk (Point blk)) CodecError m ByteString
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec (String -> DeserialiseFailure -> CodecError
CodecBytesFailure String
"BlockFetch") (Codec (BlockFetch blk (Point blk)) DeserialiseFailure m ByteString
 -> Codec (BlockFetch blk (Point blk)) CodecError m ByteString)
-> Codec
     (BlockFetch blk (Point blk)) DeserialiseFailure m ByteString
-> Codec (BlockFetch blk (Point blk)) CodecError m ByteString
forall a b. (a -> b) -> a -> b
$
              Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
-> Codec
     (BlockFetch blk (Point blk)) DeserialiseFailure m ByteString
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec (BlockFetch blk (Point blk)) e m bBF
NTN.cBlockFetchCodec Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
binaryProtocolCodecs
        , cBlockFetchCodecSerialised :: Codec
  (BlockFetch (Serialised blk) (Point blk)) CodecError m ByteString
cBlockFetchCodecSerialised =
            (DeserialiseFailure -> CodecError)
-> Codec
     (BlockFetch (Serialised blk) (Point blk))
     DeserialiseFailure
     m
     ByteString
-> Codec
     (BlockFetch (Serialised blk) (Point blk)) CodecError m ByteString
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec (String -> DeserialiseFailure -> CodecError
CodecBytesFailure String
"BlockFetchSerialised") (Codec
   (BlockFetch (Serialised blk) (Point blk))
   DeserialiseFailure
   m
   ByteString
 -> Codec
      (BlockFetch (Serialised blk) (Point blk)) CodecError m ByteString)
-> Codec
     (BlockFetch (Serialised blk) (Point blk))
     DeserialiseFailure
     m
     ByteString
-> Codec
     (BlockFetch (Serialised blk) (Point blk)) CodecError m ByteString
forall a b. (a -> b) -> a -> b
$
              Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
-> Codec
     (BlockFetch (Serialised blk) (Point blk))
     DeserialiseFailure
     m
     ByteString
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec (BlockFetch (Serialised blk) (Point blk)) e m bSBF
NTN.cBlockFetchCodecSerialised Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
binaryProtocolCodecs
        , cTxSubmission2Codec :: Codec
  (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
  CodecError
  m
  (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
cTxSubmission2Codec =
            (CodecFailure -> CodecError)
-> Codec
     (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
     CodecFailure
     m
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
-> Codec
     (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
     CodecError
     m
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec CodecFailure -> CodecError
CodecIdFailure (Codec
   (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
   CodecFailure
   m
   (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
 -> Codec
      (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
      CodecError
      m
      (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))))
-> Codec
     (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
     CodecFailure
     m
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
-> Codec
     (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
     CodecError
     m
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
forall a b. (a -> b) -> a -> b
$
              Codecs
  blk
  Any
  CodecFailure
  m
  (AnyMessage (ChainSync (Header blk) (Point blk) (Tip blk)))
  (AnyMessage
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk)))
  (AnyMessage (BlockFetch blk (Point blk)))
  (AnyMessage (BlockFetch (Serialised blk) (Point blk)))
  (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing Any))
-> Codec
     (TxSubmission2 (TxId (GenTx blk)) (GenTx blk))
     CodecFailure
     m
     (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec (TxSubmission2 (GenTxId blk) (GenTx blk)) e m bTX
NTN.cTxSubmission2Codec Codecs
  blk
  Any
  CodecFailure
  m
  (AnyMessage (ChainSync (Header blk) (Point blk) (Tip blk)))
  (AnyMessage
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk)))
  (AnyMessage (BlockFetch blk (Point blk)))
  (AnyMessage (BlockFetch (Serialised blk) (Point blk)))
  (AnyMessage (TxSubmission2 (TxId (GenTx blk)) (GenTx blk)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing Any))
forall (m :: * -> *) blk addr.
Monad m =>
Codecs
  blk
  addr
  CodecFailure
  m
  (AnyMessage (ChainSync (Header blk) (Point blk) (Tip blk)))
  (AnyMessage
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk)))
  (AnyMessage (BlockFetch blk (Point blk)))
  (AnyMessage (BlockFetch (Serialised blk) (Point blk)))
  (AnyMessage (TxSubmission2 (GenTxId blk) (GenTx blk)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing addr))
NTN.identityCodecs
        , cKeepAliveCodec :: Codec KeepAlive CodecError m (AnyMessage KeepAlive)
cKeepAliveCodec =
            (CodecFailure -> CodecError)
-> Codec KeepAlive CodecFailure m (AnyMessage KeepAlive)
-> Codec KeepAlive CodecError m (AnyMessage KeepAlive)
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec CodecFailure -> CodecError
CodecIdFailure (Codec KeepAlive CodecFailure m (AnyMessage KeepAlive)
 -> Codec KeepAlive CodecError m (AnyMessage KeepAlive))
-> Codec KeepAlive CodecFailure m (AnyMessage KeepAlive)
-> Codec KeepAlive CodecError m (AnyMessage KeepAlive)
forall a b. (a -> b) -> a -> b
$
              Codecs
  Any
  Any
  CodecFailure
  m
  (AnyMessage (ChainSync (Header Any) (Point Any) (Tip Any)))
  (AnyMessage
     (ChainSync (SerialisedHeader Any) (Point Any) (Tip Any)))
  (AnyMessage (BlockFetch Any (Point Any)))
  (AnyMessage (BlockFetch (Serialised Any) (Point Any)))
  (AnyMessage (TxSubmission2 (GenTxId Any) (GenTx Any)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing Any))
-> Codec KeepAlive CodecFailure m (AnyMessage KeepAlive)
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec KeepAlive e m bKA
NTN.cKeepAliveCodec Codecs
  Any
  Any
  CodecFailure
  m
  (AnyMessage (ChainSync (Header Any) (Point Any) (Tip Any)))
  (AnyMessage
     (ChainSync (SerialisedHeader Any) (Point Any) (Tip Any)))
  (AnyMessage (BlockFetch Any (Point Any)))
  (AnyMessage (BlockFetch (Serialised Any) (Point Any)))
  (AnyMessage (TxSubmission2 (GenTxId Any) (GenTx Any)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing Any))
forall (m :: * -> *) blk addr.
Monad m =>
Codecs
  blk
  addr
  CodecFailure
  m
  (AnyMessage (ChainSync (Header blk) (Point blk) (Tip blk)))
  (AnyMessage
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk)))
  (AnyMessage (BlockFetch blk (Point blk)))
  (AnyMessage (BlockFetch (Serialised blk) (Point blk)))
  (AnyMessage (TxSubmission2 (GenTxId blk) (GenTx blk)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing addr))
NTN.identityCodecs
        , cPeerSharingCodec :: Codec
  (PeerSharing NodeId) CodecError m (AnyMessage (PeerSharing NodeId))
cPeerSharingCodec =
            (CodecFailure -> CodecError)
-> Codec
     (PeerSharing NodeId)
     CodecFailure
     m
     (AnyMessage (PeerSharing NodeId))
-> Codec
     (PeerSharing NodeId) CodecError m (AnyMessage (PeerSharing NodeId))
forall (m :: * -> *) failure failure' ps bytes.
Functor m =>
(failure -> failure')
-> Codec ps failure m bytes -> Codec ps failure' m bytes
mapFailureCodec CodecFailure -> CodecError
CodecIdFailure (Codec
   (PeerSharing NodeId)
   CodecFailure
   m
   (AnyMessage (PeerSharing NodeId))
 -> Codec
      (PeerSharing NodeId)
      CodecError
      m
      (AnyMessage (PeerSharing NodeId)))
-> Codec
     (PeerSharing NodeId)
     CodecFailure
     m
     (AnyMessage (PeerSharing NodeId))
-> Codec
     (PeerSharing NodeId) CodecError m (AnyMessage (PeerSharing NodeId))
forall a b. (a -> b) -> a -> b
$
              Codecs
  Any
  NodeId
  CodecFailure
  m
  (AnyMessage (ChainSync (Header Any) (Point Any) (Tip Any)))
  (AnyMessage
     (ChainSync (SerialisedHeader Any) (Point Any) (Tip Any)))
  (AnyMessage (BlockFetch Any (Point Any)))
  (AnyMessage (BlockFetch (Serialised Any) (Point Any)))
  (AnyMessage (TxSubmission2 (GenTxId Any) (GenTx Any)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing NodeId))
-> Codec
     (PeerSharing NodeId)
     CodecFailure
     m
     (AnyMessage (PeerSharing NodeId))
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec (PeerSharing addr) e m bPS
NTN.cPeerSharingCodec Codecs
  Any
  NodeId
  CodecFailure
  m
  (AnyMessage (ChainSync (Header Any) (Point Any) (Tip Any)))
  (AnyMessage
     (ChainSync (SerialisedHeader Any) (Point Any) (Tip Any)))
  (AnyMessage (BlockFetch Any (Point Any)))
  (AnyMessage (BlockFetch (Serialised Any) (Point Any)))
  (AnyMessage (TxSubmission2 (GenTxId Any) (GenTx Any)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing NodeId))
forall (m :: * -> *) blk addr.
Monad m =>
Codecs
  blk
  addr
  CodecFailure
  m
  (AnyMessage (ChainSync (Header blk) (Point blk) (Tip blk)))
  (AnyMessage
     (ChainSync (SerialisedHeader blk) (Point blk) (Tip blk)))
  (AnyMessage (BlockFetch blk (Point blk)))
  (AnyMessage (BlockFetch (Serialised blk) (Point blk)))
  (AnyMessage (TxSubmission2 (GenTxId blk) (GenTx blk)))
  (AnyMessage KeepAlive)
  (AnyMessage (PeerSharing addr))
NTN.identityCodecs
        }
      where
        binaryProtocolCodecs :: Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
binaryProtocolCodecs = CodecConfig blk
-> BlockNodeToNodeVersion blk
-> (NodeToNodeVersion -> NodeId -> Encoding)
-> (NodeToNodeVersion -> forall s. Decoder s NodeId)
-> NodeToNodeVersion
-> Codecs
     blk
     NodeId
     DeserialiseFailure
     m
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
forall (m :: * -> *) blk addr.
(IOLike m, SerialiseNodeToNodeConstraints blk) =>
CodecConfig blk
-> BlockNodeToNodeVersion blk
-> (NodeToNodeVersion -> addr -> Encoding)
-> (NodeToNodeVersion -> forall s. Decoder s addr)
-> NodeToNodeVersion
-> Codecs
     blk
     addr
     DeserialiseFailure
     m
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
NTN.defaultCodecs (TopLevelConfig blk -> CodecConfig blk
forall blk. TopLevelConfig blk -> CodecConfig blk
configCodec TopLevelConfig blk
cfg) BlockNodeToNodeVersion blk
blockVersion ((NodeId -> Encoding) -> NodeToNodeVersion -> NodeId -> Encoding
forall a b. a -> b -> a
const NodeId -> Encoding
encodeNodeId) (Decoder s NodeId -> NodeToNodeVersion -> Decoder s NodeId
forall a b. a -> b -> a
const Decoder s NodeId
forall s. Decoder s NodeId
decodeNodeId) NodeToNodeVersion
ntnVersion

-- | Sum of 'CodecFailure' (from @identityCodecs@) and 'DeserialiseFailure'
-- (from @defaultCodecs@).
data CodecError
  = CodecIdFailure CodecFailure
  | CodecBytesFailure
      String  -- ^ Extra error message, e.g., the name of the codec
      DeserialiseFailure
  deriving (Int -> CodecError -> ShowS
[CodecError] -> ShowS
CodecError -> String
(Int -> CodecError -> ShowS)
-> (CodecError -> String)
-> ([CodecError] -> ShowS)
-> Show CodecError
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> CodecError -> ShowS
showsPrec :: Int -> CodecError -> ShowS
$cshow :: CodecError -> String
show :: CodecError -> String
$cshowList :: [CodecError] -> ShowS
showList :: [CodecError] -> ShowS
Show, Show CodecError
Typeable CodecError
(Typeable CodecError, Show CodecError) =>
(CodecError -> SomeException)
-> (SomeException -> Maybe CodecError)
-> (CodecError -> String)
-> (CodecError -> Bool)
-> Exception CodecError
SomeException -> Maybe CodecError
CodecError -> Bool
CodecError -> String
CodecError -> SomeException
forall e.
(Typeable e, Show e) =>
(e -> SomeException)
-> (SomeException -> Maybe e)
-> (e -> String)
-> (e -> Bool)
-> Exception e
$ctoException :: CodecError -> SomeException
toException :: CodecError -> SomeException
$cfromException :: SomeException -> Maybe CodecError
fromException :: SomeException -> Maybe CodecError
$cdisplayException :: CodecError -> String
displayException :: CodecError -> String
$cbacktraceDesired :: CodecError -> Bool
backtraceDesired :: CodecError -> Bool
Exception)

{-------------------------------------------------------------------------------
  Running an edge
-------------------------------------------------------------------------------}

-- | Cause for an edge to restart
--
data RestartCause
  = RestartScheduled
    -- ^ restart because at least one of the two nodes set its status to
    -- 'VFalling' because of a scheduled restart in 'tnaRestarts'
  | RestartChainSyncTerminated
    -- ^ restart because the ChainSync client terminated the mini protocol

-- | Fork two directed edges, one in each direction between the two vertices
--
forkBothEdges ::
     (IOLike m, RunNode blk, HasCallStack)
  => ResourceRegistry m
  -> OracularClock m
  -> Tracer m (SlotNo, MiniProtocolState)
  -> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
  -> (CodecConfig blk, CalcMessageDelay blk)
  -> Map CoreNodeId (VertexStatusVar m blk)
  -> (CoreNodeId, CoreNodeId)
  -> m [((CoreNodeId, CoreNodeId), EdgeStatusVar m)]
forkBothEdges :: forall (m :: * -> *) blk.
(IOLike m, RunNode blk, HasCallStack) =>
ResourceRegistry m
-> OracularClock m
-> Tracer m (SlotNo, MiniProtocolState)
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> Map CoreNodeId (VertexStatusVar m blk)
-> (CoreNodeId, CoreNodeId)
-> m [((CoreNodeId, CoreNodeId), EdgeStatusVar m)]
forkBothEdges ResourceRegistry m
sharedRegistry OracularClock m
clock Tracer m (SlotNo, MiniProtocolState)
tr (NodeToNodeVersion, BlockNodeToNodeVersion blk)
version (CodecConfig blk, CalcMessageDelay blk)
cfg Map CoreNodeId (VertexStatusVar m blk)
vertexStatusVars (CoreNodeId
node1, CoreNodeId
node2) = do
  let endpoint1 :: (CoreNodeId, VertexStatusVar m blk)
endpoint1 = CoreNodeId -> (CoreNodeId, VertexStatusVar m blk)
mkEndpoint CoreNodeId
node1
      endpoint2 :: (CoreNodeId, VertexStatusVar m blk)
endpoint2 = CoreNodeId -> (CoreNodeId, VertexStatusVar m blk)
mkEndpoint CoreNodeId
node2
      mkEndpoint :: CoreNodeId -> (CoreNodeId, VertexStatusVar m blk)
mkEndpoint CoreNodeId
node = case CoreNodeId
-> Map CoreNodeId (VertexStatusVar m blk)
-> Maybe (VertexStatusVar m blk)
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup CoreNodeId
node Map CoreNodeId (VertexStatusVar m blk)
vertexStatusVars of
          Maybe (VertexStatusVar m blk)
Nothing  -> String -> (CoreNodeId, VertexStatusVar m blk)
forall a. HasCallStack => String -> a
error (String -> (CoreNodeId, VertexStatusVar m blk))
-> String -> (CoreNodeId, VertexStatusVar m blk)
forall a b. (a -> b) -> a -> b
$ String
"node not found: " String -> ShowS
forall a. [a] -> [a] -> [a]
++ CoreNodeId -> String
forall a. Show a => a -> String
show CoreNodeId
node
          Just VertexStatusVar m blk
var -> (CoreNodeId
node, VertexStatusVar m blk
var)

  let mkDirEdge :: (CoreNodeId, VertexStatusVar m blk)
-> (CoreNodeId, VertexStatusVar m blk)
-> m ((CoreNodeId, CoreNodeId), StrictTVar m EdgeStatus)
mkDirEdge (CoreNodeId, VertexStatusVar m blk)
e1 (CoreNodeId, VertexStatusVar m blk)
e2 = do
        v <- EdgeStatus -> m (StrictTVar m EdgeStatus)
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
uncheckedNewTVarM EdgeStatus
EDown
        let label = [String] -> String
forall (t :: * -> *) a. Foldable t => t [a] -> [a]
concat
              [String
"directed-edge-", CoreNodeId -> String
forall a. Condense a => a -> String
condense ((CoreNodeId, VertexStatusVar m blk) -> CoreNodeId
forall a b. (a, b) -> a
fst (CoreNodeId, VertexStatusVar m blk)
e1), String
"-", CoreNodeId -> String
forall a. Condense a => a -> String
condense ((CoreNodeId, VertexStatusVar m blk) -> CoreNodeId
forall a b. (a, b) -> a
fst (CoreNodeId, VertexStatusVar m blk)
e2)]
        void $ forkLinkedThread sharedRegistry label $ do
          directedEdge sharedRegistry tr version cfg clock v e1 e2
        pure ((fst e1, fst e2), v)

  ev12 <- (CoreNodeId, VertexStatusVar m blk)
-> (CoreNodeId, VertexStatusVar m blk)
-> m ((CoreNodeId, CoreNodeId), StrictTVar m EdgeStatus)
mkDirEdge (CoreNodeId, VertexStatusVar m blk)
endpoint1 (CoreNodeId, VertexStatusVar m blk)
endpoint2
  ev21 <- mkDirEdge endpoint2 endpoint1

  pure [ev12, ev21]

-- | Spawn all mini protocols' threads for a given directed edge in the node
-- network topology (ie an ordered pair of core nodes, with client first and
-- server second)
--
-- The edge cannot start until both nodes are simultaneously 'VUp'.
--
-- The edge may restart itself for the reasons modeled by 'RestartCause'
--
-- The actual control flow here does not faithfully model the real
-- implementation. On an exception, for example, the actual node implementation
-- kills the other threads on the same peer as the thread that threw the
-- exception, and then relies on TCP socket semantics to eventually kill the
-- corresponding threads on the remote peer. The client node recreates its
-- client threads after a delay, and they reconnect to the remote peer, thereby
-- recreating the server threads.
--
-- This model instead propagates the exception to the rest of the /un/directed
-- edge via the @async@ interface rather than relying on some sort of mock
-- socket semantics to convey the cancellation.
directedEdge ::
  forall m blk. (IOLike m, RunNode blk)
  => ResourceRegistry m
  -> Tracer m (SlotNo, MiniProtocolState)
  -> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
  -> (CodecConfig blk, CalcMessageDelay blk)
  -> OracularClock m
  -> EdgeStatusVar m
  -> (CoreNodeId, VertexStatusVar m blk)
  -> (CoreNodeId, VertexStatusVar m blk)
  -> m ()
directedEdge :: forall (m :: * -> *) blk.
(IOLike m, RunNode blk) =>
ResourceRegistry m
-> Tracer m (SlotNo, MiniProtocolState)
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> OracularClock m
-> EdgeStatusVar m
-> (CoreNodeId, VertexStatusVar m blk)
-> (CoreNodeId, VertexStatusVar m blk)
-> m ()
directedEdge ResourceRegistry m
registry Tracer m (SlotNo, MiniProtocolState)
tr (NodeToNodeVersion, BlockNodeToNodeVersion blk)
version (CodecConfig blk, CalcMessageDelay blk)
cfg OracularClock m
clock EdgeStatusVar m
edgeStatusVar (CoreNodeId, VertexStatusVar m blk)
client (CoreNodeId, VertexStatusVar m blk)
server =
    m ()
loop
  where
    loop :: m ()
loop = do
        restart <- ResourceRegistry m
-> OracularClock m
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> EdgeStatusVar m
-> (CoreNodeId, VertexStatusVar m blk)
-> (CoreNodeId, VertexStatusVar m blk)
-> m RestartCause
forall (m :: * -> *) blk.
(IOLike m, RunNode blk) =>
ResourceRegistry m
-> OracularClock m
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> EdgeStatusVar m
-> (CoreNodeId, VertexStatusVar m blk)
-> (CoreNodeId, VertexStatusVar m blk)
-> m RestartCause
directedEdgeInner ResourceRegistry m
registry OracularClock m
clock (NodeToNodeVersion, BlockNodeToNodeVersion blk)
version (CodecConfig blk, CalcMessageDelay blk)
cfg EdgeStatusVar m
edgeStatusVar (CoreNodeId, VertexStatusVar m blk)
client (CoreNodeId, VertexStatusVar m blk)
server
          m RestartCause
-> (SomeException -> m RestartCause) -> m RestartCause
forall e a. Exception e => m a -> (e -> m a) -> m a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` SomeException -> m RestartCause
forall a. SomeException -> m a
hUnexpected
        atomically $ writeTVar edgeStatusVar EDown
        case restart of
          RestartCause
RestartScheduled            -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          RestartCause
RestartChainSyncTerminated  -> do
            -- "error" policy: restart at beginning of next slot
            s <- OracularClock m -> m SlotNo
forall (m :: * -> *). OracularClock m -> m SlotNo
OracularClock.getCurrentSlot OracularClock m
clock
            let s' = SlotNo -> SlotNo
forall a. Enum a => a -> a
succ SlotNo
s
            traceWith tr (s, MiniProtocolDelayed)
            void $ OracularClock.blockUntilSlot clock s'
            traceWith tr (s', MiniProtocolRestarting)
        loop
      where
        -- Wrap synchronous exceptions in 'MiniProtocolFatalException'
        --
        hUnexpected :: forall a. SomeException -> m a
        hUnexpected :: forall a. SomeException -> m a
hUnexpected e :: SomeException
e@(Exn.SomeException e
e') = case SomeException -> Maybe AsyncException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e of
          Just (AsyncException
_ :: Exn.AsyncException) -> SomeException -> m a
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO SomeException
e
          Maybe AsyncException
Nothing                        -> case SomeException -> Maybe SomeAsyncException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e of
            Just (SomeAsyncException
_ :: Exn.SomeAsyncException) -> SomeException -> m a
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO SomeException
e
            Maybe SomeAsyncException
Nothing                            -> MiniProtocolFatalException -> m a
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO MiniProtocolFatalException
              { mpfeType :: TypeRep
mpfeType   = e -> TypeRep
forall a. Typeable a => a -> TypeRep
Typeable.typeOf e
e'
              , mpfeExn :: SomeException
mpfeExn    = SomeException
e
              , mpfeClient :: CoreNodeId
mpfeClient = (CoreNodeId, VertexStatusVar m blk) -> CoreNodeId
forall a b. (a, b) -> a
fst (CoreNodeId, VertexStatusVar m blk)
client
              , mpfeServer :: CoreNodeId
mpfeServer = (CoreNodeId, VertexStatusVar m blk) -> CoreNodeId
forall a b. (a, b) -> a
fst (CoreNodeId, VertexStatusVar m blk)
server
              }

-- | Spawn threads for all of the mini protocols
--
-- See 'directedEdge'.
directedEdgeInner ::
  forall m blk. (IOLike m, RunNode blk)
  => ResourceRegistry m
  -> OracularClock m
  -> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
  -> (CodecConfig blk, CalcMessageDelay blk)
  -> EdgeStatusVar m
  -> (CoreNodeId, VertexStatusVar m blk)
     -- ^ client threads on this node
  -> (CoreNodeId, VertexStatusVar m blk)
     -- ^ server threads on this node
  -> m RestartCause
directedEdgeInner :: forall (m :: * -> *) blk.
(IOLike m, RunNode blk) =>
ResourceRegistry m
-> OracularClock m
-> (NodeToNodeVersion, BlockNodeToNodeVersion blk)
-> (CodecConfig blk, CalcMessageDelay blk)
-> EdgeStatusVar m
-> (CoreNodeId, VertexStatusVar m blk)
-> (CoreNodeId, VertexStatusVar m blk)
-> m RestartCause
directedEdgeInner ResourceRegistry m
registry OracularClock m
clock (NodeToNodeVersion
version, BlockNodeToNodeVersion blk
blockVersion) (CodecConfig blk
cfg, CalcMessageDelay blk
calcMessageDelay) EdgeStatusVar m
edgeStatusVar
  (CoreNodeId
node1, VertexStatusVar m blk
vertexStatusVar1) (CoreNodeId
node2, VertexStatusVar m blk
vertexStatusVar2) = do
    -- block until both nodes are 'VUp'
    (LimitedApp app1, LimitedApp app2) <- STM m (LimitedApp m NodeId blk, LimitedApp m NodeId blk)
-> m (LimitedApp m NodeId blk, LimitedApp m NodeId blk)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (LimitedApp m NodeId blk, LimitedApp m NodeId blk)
 -> m (LimitedApp m NodeId blk, LimitedApp m NodeId blk))
-> STM m (LimitedApp m NodeId blk, LimitedApp m NodeId blk)
-> m (LimitedApp m NodeId blk, LimitedApp m NodeId blk)
forall a b. (a -> b) -> a -> b
$ do
      (,) (LimitedApp m NodeId blk
 -> LimitedApp m NodeId blk
 -> (LimitedApp m NodeId blk, LimitedApp m NodeId blk))
-> STM m (LimitedApp m NodeId blk)
-> STM
     m
     (LimitedApp m NodeId blk
      -> (LimitedApp m NodeId blk, LimitedApp m NodeId blk))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> VertexStatusVar m blk -> STM m (LimitedApp m NodeId blk)
forall {m :: * -> *} {m :: * -> *} {blk}.
MonadSTM m =>
StrictTVar m (VertexStatus m blk)
-> STM m (LimitedApp m NodeId blk)
getApp VertexStatusVar m blk
vertexStatusVar1 STM
  m
  (LimitedApp m NodeId blk
   -> (LimitedApp m NodeId blk, LimitedApp m NodeId blk))
-> STM m (LimitedApp m NodeId blk)
-> STM m (LimitedApp m NodeId blk, LimitedApp m NodeId blk)
forall a b. STM m (a -> b) -> STM m a -> STM m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> VertexStatusVar m blk -> STM m (LimitedApp m NodeId blk)
forall {m :: * -> *} {m :: * -> *} {blk}.
MonadSTM m =>
StrictTVar m (VertexStatus m blk)
-> STM m (LimitedApp m NodeId blk)
getApp VertexStatusVar m blk
vertexStatusVar2

    atomically $ writeTVar edgeStatusVar EUp

    let miniProtocol ::
             String
             -- ^ protocol name
          -> (String -> a -> RestartCause)
          -> (String -> b -> RestartCause)
          -> (  LimitedApp' m NodeId blk
             -> NodeToNodeVersion
             -> ExpandedInitiatorContext NodeId m
             -> Channel m msg
             -> m (a, trailingBytes)
             )
            -- ^ client action to run on node1
          -> (  LimitedApp' m NodeId blk
             -> NodeToNodeVersion
             -> ResponderContext NodeId
             -> Channel m msg
             -> m (b, trailingBytes)
             )
             -- ^ server action to run on node2
          -> (msg -> m ())
          -> m (m RestartCause, m RestartCause)
        miniProtocol String
proto String -> a -> RestartCause
retClient String -> b -> RestartCause
retServer LimitedApp' m NodeId blk
-> NodeToNodeVersion
-> ExpandedInitiatorContext NodeId m
-> Channel m msg
-> m (a, trailingBytes)
client LimitedApp' m NodeId blk
-> NodeToNodeVersion
-> ResponderContext NodeId
-> Channel m msg
-> m (b, trailingBytes)
server msg -> m ()
middle = do
             (chan, dualChan) <-
               ResourceRegistry m
-> (CoreNodeId, CoreNodeId, String)
-> (msg -> m ())
-> m (Channel m msg, Channel m msg)
forall (m :: * -> *) a.
IOLike m =>
ResourceRegistry m
-> (CoreNodeId, CoreNodeId, String)
-> (a -> m ())
-> m (Channel m a, Channel m a)
createConnectedChannelsWithDelay ResourceRegistry m
registry (CoreNodeId
node1, CoreNodeId
node2, String
proto) msg -> m ()
middle
             pure
               ( (retClient (proto <> ".client") . fst) <$> client app1 version initiatorCtx chan
               , (retServer (proto <> ".server") . fst) <$> server app2 version responderCtx dualChan
               )
          where
            initiatorCtx :: ExpandedInitiatorContext NodeId m
initiatorCtx = ExpandedInitiatorContext {
                eicConnectionId :: ConnectionId NodeId
eicConnectionId    = NodeId -> NodeId -> ConnectionId NodeId
forall addr. addr -> addr -> ConnectionId addr
ConnectionId (CoreNodeId -> NodeId
fromCoreNodeId CoreNodeId
node1) (CoreNodeId -> NodeId
fromCoreNodeId CoreNodeId
node2),
                eicControlMessage :: ControlMessageSTM m
eicControlMessage  = ControlMessage -> ControlMessageSTM m
forall a. a -> STM m a
forall (m :: * -> *) a. Monad m => a -> m a
return ControlMessage
Continue,
                eicIsBigLedgerPeer :: IsBigLedgerPeer
eicIsBigLedgerPeer = IsBigLedgerPeer
IsNotBigLedgerPeer
              }
            responderCtx :: ResponderContext NodeId
responderCtx = ResponderContext {
                rcConnectionId :: ConnectionId NodeId
rcConnectionId     = NodeId -> NodeId -> ConnectionId NodeId
forall addr. addr -> addr -> ConnectionId addr
ConnectionId (CoreNodeId -> NodeId
fromCoreNodeId CoreNodeId
node1) (CoreNodeId -> NodeId
fromCoreNodeId CoreNodeId
node2)
              }

    (>>= withAsyncsWaitAny) $
      fmap flattenPairs $
      sequence $
        pure (watcher vertexStatusVar1, watcher vertexStatusVar2)
        NE.:|
      [ miniProtocol "ChainSync"
          (\String
_s NodeToNodeInitiatorResult
_  -> RestartCause
RestartChainSyncTerminated)
          (\String
_s () -> RestartCause
RestartChainSyncTerminated)
          NTN.aChainSyncClient
          NTN.aChainSyncServer
          chainSyncMiddle
      , miniProtocol "BlockFetch"
          neverReturns
          neverReturns
          NTN.aBlockFetchClient
          NTN.aBlockFetchServer
          (\ByteString
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
      , miniProtocol "TxSubmission"
          neverReturns
          neverReturns
          NTN.aTxSubmission2Client
          NTN.aTxSubmission2Server
          (\AnyMessage (TxSubmission2 (GenTxId blk) (GenTx blk))
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
      , miniProtocol "KeepAlive"
          neverReturns
          neverReturns
          NTN.aKeepAliveClient
          NTN.aKeepAliveServer
          (\AnyMessage KeepAlive
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
      ]
  where
    getApp :: StrictTVar m (VertexStatus m blk)
-> STM m (LimitedApp m NodeId blk)
getApp StrictTVar m (VertexStatus m blk)
v = StrictTVar m (VertexStatus m blk) -> STM m (VertexStatus m blk)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (VertexStatus m blk)
v STM m (VertexStatus m blk)
-> (VertexStatus m blk -> STM m (LimitedApp m NodeId blk))
-> STM m (LimitedApp m NodeId blk)
forall a b. STM m a -> (a -> STM m b) -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      VUp NodeKernel m NodeId Void blk
_ LimitedApp m NodeId blk
app -> LimitedApp m NodeId blk -> STM m (LimitedApp m NodeId blk)
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure LimitedApp m NodeId blk
app
      VertexStatus m blk
_         -> STM m (LimitedApp m NodeId blk)
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry

    flattenPairs :: forall a. NE.NonEmpty (a, a) -> NE.NonEmpty a
    flattenPairs :: forall a. NonEmpty (a, a) -> NonEmpty a
flattenPairs = (NonEmpty a -> NonEmpty a -> NonEmpty a)
-> (NonEmpty a, NonEmpty a) -> NonEmpty a
forall a b c. (a -> b -> c) -> (a, b) -> c
uncurry NonEmpty a -> NonEmpty a -> NonEmpty a
forall a. Semigroup a => a -> a -> a
(<>) ((NonEmpty a, NonEmpty a) -> NonEmpty a)
-> (NonEmpty (a, a) -> (NonEmpty a, NonEmpty a))
-> NonEmpty (a, a)
-> NonEmpty a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NonEmpty (a, a) -> (NonEmpty a, NonEmpty a)
forall (f :: * -> *) a b. Functor f => f (a, b) -> (f a, f b)
neUnzip

    neverReturns :: forall x void. String -> x -> void
    neverReturns :: forall x void. String -> x -> void
neverReturns String
s !x
_ = String -> void
forall a. HasCallStack => String -> a
error (String -> void) -> String -> void
forall a b. (a -> b) -> a -> b
$ String
s String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" never returns!"

    -- terminates (by returning, not via exception) when the vertex starts
    -- 'VFalling'
    --
    -- because of 'withAsyncsWaitAny' used above, this brings down the whole
    -- edge
    watcher :: VertexStatusVar m blk -> m RestartCause
    watcher :: VertexStatusVar m blk -> m RestartCause
watcher VertexStatusVar m blk
v = do
        STM m RestartCause -> m RestartCause
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m RestartCause -> m RestartCause)
-> STM m RestartCause -> m RestartCause
forall a b. (a -> b) -> a -> b
$ VertexStatusVar m blk -> STM m (VertexStatus m blk)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar VertexStatusVar m blk
v STM m (VertexStatus m blk)
-> (VertexStatus m blk -> STM m RestartCause) -> STM m RestartCause
forall a b. STM m a -> (a -> STM m b) -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          VertexStatus m blk
VFalling -> RestartCause -> STM m RestartCause
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure RestartCause
RestartScheduled
          VertexStatus m blk
_        -> STM m RestartCause
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry

    -- introduce a delay for 'CS.MsgRollForward'
    --
    -- It is reasonable to delay only this message because this message is the
    -- first step in process of one node diffusing a block to another node.
    chainSyncMiddle :: Lazy.ByteString -> m ()
    chainSyncMiddle :: ByteString -> m ()
chainSyncMiddle ByteString
bs = do
        let tok :: SingChainSync ('StNext 'StMustReply)
tok = SingNextKind 'StMustReply -> SingChainSync ('StNext 'StMustReply)
forall {header} {point} {tip} (k1 :: StNextKind).
SingNextKind k1 -> SingChainSync ('StNext k1)
CS.SingNext SingNextKind 'StMustReply
CS.SingMustReply
        decodeStep :: Codec.DecodeStep
                        Lazy.ByteString DeserialiseFailure m
                        (Codec.SomeMessage ('CS.StNext 'CS.StMustReply))
          <- Codec
  (ChainSync (Header blk) (Point blk) (Tip blk))
  DeserialiseFailure
  m
  ByteString
-> forall (st :: ChainSync (Header blk) (Point blk) (Tip blk)).
   ActiveState st =>
   StateToken st
   -> m (DecodeStep ByteString DeserialiseFailure m (SomeMessage st))
forall ps failure (m :: * -> *) bytes.
Codec ps failure m bytes
-> forall (st :: ps).
   ActiveState st =>
   StateToken st -> m (DecodeStep bytes failure m (SomeMessage st))
Codec.decode Codec
  (ChainSync (Header blk) (Point blk) (Tip blk))
  DeserialiseFailure
  m
  ByteString
codec SingChainSync ('StNext 'StMustReply)
StateToken ('StNext 'StMustReply)
forall {header} {point} {tip}. SingChainSync ('StNext 'StMustReply)
tok
        Codec.runDecoder [bs] decodeStep >>= \case
          Right (Codec.SomeMessage (CS.MsgRollForward Header blk
hdr Tip blk
_tip)) -> do
              s <- OracularClock m -> m SlotNo
forall (m :: * -> *). OracularClock m -> m SlotNo
OracularClock.getCurrentSlot OracularClock m
clock
              let NumSlots d = f (node1, node2) s hdr
                    where
                      CalcMessageDelay f = calcMessageDelay
              void $ OracularClock.blockUntilSlot clock $ blockSlot hdr + SlotNo d
          Either DeserialiseFailure (SomeMessage ('StNext 'StMustReply))
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      where
        codec :: Codec
  (ChainSync (Header blk) (Point blk) (Tip blk))
  DeserialiseFailure
  m
  ByteString
codec =
            Codecs
  blk
  NodeId
  DeserialiseFailure
  m
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
  ByteString
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
forall blk addr e (m :: * -> *) bCS bSCS bBF bSBF bTX bKA bPS.
Codecs blk addr e m bCS bSCS bBF bSBF bTX bKA bPS
-> Codec (ChainSync (Header blk) (Point blk) (Tip blk)) e m bCS
NTN.cChainSyncCodec (Codecs
   blk
   NodeId
   DeserialiseFailure
   m
   ByteString
   ByteString
   ByteString
   ByteString
   ByteString
   ByteString
   ByteString
 -> Codec
      (ChainSync (Header blk) (Point blk) (Tip blk))
      DeserialiseFailure
      m
      ByteString)
-> Codecs
     blk
     NodeId
     DeserialiseFailure
     m
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
-> Codec
     (ChainSync (Header blk) (Point blk) (Tip blk))
     DeserialiseFailure
     m
     ByteString
forall a b. (a -> b) -> a -> b
$ CodecConfig blk
-> BlockNodeToNodeVersion blk
-> (NodeToNodeVersion -> NodeId -> Encoding)
-> (NodeToNodeVersion -> forall s. Decoder s NodeId)
-> NodeToNodeVersion
-> Codecs
     blk
     NodeId
     DeserialiseFailure
     m
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
forall (m :: * -> *) blk addr.
(IOLike m, SerialiseNodeToNodeConstraints blk) =>
CodecConfig blk
-> BlockNodeToNodeVersion blk
-> (NodeToNodeVersion -> addr -> Encoding)
-> (NodeToNodeVersion -> forall s. Decoder s addr)
-> NodeToNodeVersion
-> Codecs
     blk
     addr
     DeserialiseFailure
     m
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
     ByteString
NTN.defaultCodecs CodecConfig blk
cfg BlockNodeToNodeVersion blk
blockVersion ((NodeId -> Encoding) -> NodeToNodeVersion -> NodeId -> Encoding
forall a b. a -> b -> a
const NodeId -> Encoding
encodeNodeId) (Decoder s NodeId -> NodeToNodeVersion -> Decoder s NodeId
forall a b. a -> b -> a
const Decoder s NodeId
forall s. Decoder s NodeId
decodeNodeId) NodeToNodeVersion
version

-- | Variant of 'createConnectChannels' with intermediate queues for
-- delayed-but-in-order messages
--
-- Sending adds the message to a queue. The delaying argument should use
-- 'threadDelay' in order to delay the transfer of the given message from the
-- queue to the recipient.
createConnectedChannelsWithDelay ::
     IOLike m
  => ResourceRegistry m
  -> (CoreNodeId, CoreNodeId, String)
     -- ^ (client, server, protocol)
  -> (a -> m ())
     -- ^ per-message delay
  -> m (Channel m a, Channel m a)
createConnectedChannelsWithDelay :: forall (m :: * -> *) a.
IOLike m =>
ResourceRegistry m
-> (CoreNodeId, CoreNodeId, String)
-> (a -> m ())
-> m (Channel m a, Channel m a)
createConnectedChannelsWithDelay ResourceRegistry m
registry (CoreNodeId
client, CoreNodeId
server, String
proto) a -> m ()
middle = do
    -- queue for async send and an mvar for delayed-but-in-order reads from the
    -- queue
    qA <- STM m (TQueue m a) -> m (TQueue m a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (TQueue m a) -> m (TQueue m a))
-> STM m (TQueue m a) -> m (TQueue m a)
forall a b. (a -> b) -> a -> b
$ STM m (TQueue m a)
forall a. STM m (TQueue m a)
forall (m :: * -> *) a. MonadSTM m => STM m (TQueue m a)
MonadSTM.newTQueue
    bA <- atomically $ MonadSTM.newEmptyTMVar
    spawn (client, server) qA bA

    qB <- atomically $ MonadSTM.newTQueue
    bB <- atomically $ MonadSTM.newEmptyTMVar
    spawn (server, client) qB bB

    return (chan qA bB, chan qB bA)   -- note the crossover
  where
    spawn :: (CoreNodeId, CoreNodeId) -> TQueue m a -> TMVar m a -> m ()
spawn (CoreNodeId
cid1, CoreNodeId
cid2) TQueue m a
q TMVar m a
b = do
        let label :: String
label =
                String
"delaying thread for " String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
proto String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" " String -> ShowS
forall a. Semigroup a => a -> a -> a
<>
                CoreNodeId -> String
forall a. Show a => a -> String
show CoreNodeId
cid1 String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" to " String -> ShowS
forall a. Semigroup a => a -> a -> a
<> CoreNodeId -> String
forall a. Show a => a -> String
show CoreNodeId
cid2
        m (Thread m Any) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (Thread m Any) -> m ()) -> m (Thread m Any) -> m ()
forall a b. (a -> b) -> a -> b
$ 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
label (m Any -> m (Thread m Any)) -> m Any -> m (Thread m Any)
forall a b. (a -> b) -> a -> b
$ m () -> m Any
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (m () -> m Any) -> m () -> m Any
forall a b. (a -> b) -> a -> b
$ do
          x <- 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 -> m a) -> STM m a -> m a
forall a b. (a -> b) -> a -> b
$ TQueue m a -> STM m a
forall a. TQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TQueue m a -> STM m a
MonadSTM.readTQueue TQueue m a
q
          middle x
          atomically $ MonadSTM.putTMVar b x

    chan :: TQueue m a -> TMVar m a -> Channel m a
chan TQueue m a
q TMVar m a
b = Channel
        { recv :: m (Maybe a)
recv = (a -> Maybe a) -> m a -> m (Maybe a)
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap a -> Maybe a
forall a. a -> Maybe a
Just (m a -> m (Maybe a)) -> m a -> m (Maybe a)
forall a b. (a -> b) -> a -> b
$ 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 -> m a) -> STM m a -> m a
forall a b. (a -> b) -> a -> b
$ TMVar m a -> STM m a
forall a. TMVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TMVar m a -> STM m a
MonadSTM.takeTMVar TMVar m a
b
        , send :: a -> m ()
send = 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 ()) -> (a -> STM m ()) -> a -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TQueue m a -> a -> STM m ()
forall a. TQueue m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TQueue m a -> a -> STM m ()
MonadSTM.writeTQueue TQueue m a
q
        }

{-------------------------------------------------------------------------------
  Node information not bound to lifetime of a specific node instance
-------------------------------------------------------------------------------}

data NodeInfo blk db ev = NodeInfo
  { forall blk db (ev :: * -> *).
NodeInfo blk db ev -> NodeEvents blk ev
nodeInfoEvents :: NodeEvents blk ev
  , forall blk db (ev :: * -> *). NodeInfo blk db ev -> NodeDBs db
nodeInfoDBs    :: NodeDBs db
  }

-- | A vector with an @ev@-shaped element for a particular set of
-- instrumentation events
--
-- The @ev@ type parameter is instantiated by this module at types for
-- 'Tracer's and lists: actions for accumulating and lists as accumulations.
data NodeEvents blk ev = NodeEvents
  { forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, RealPoint blk, BlockNo)
nodeEventsAdds        :: ev (SlotNo, RealPoint blk, BlockNo)
    -- ^ every 'AddedBlockToVolatileDB' excluding EBBs
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (TraceForgeEvent blk)
nodeEventsForges      :: ev (TraceForgeEvent blk)
    -- ^ every 'TraceForgeEvent'
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, RealPoint blk, BlockNo)
nodeEventsHeaderAdds  :: ev (SlotNo, RealPoint blk, BlockNo)
    -- ^ every 'TraceDownloadedHeader', excluding EBBs
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (RealPoint blk, ExtValidationError blk)
nodeEventsInvalids    :: ev (RealPoint blk, ExtValidationError blk)
    -- ^ the point of every 'ChainDB.InvalidBlock' event
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, RealPoint blk, BlockNo)
nodeEventsSelects     :: ev (SlotNo, RealPoint blk, BlockNo)
    -- ^ every 'ChainDB.AddedToCurrentChain' and 'ChainDB.SwitchedToAFork'
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (SlotNo, WithOrigin BlockNo)
nodeEventsTipBlockNos :: ev (SlotNo, WithOrigin BlockNo)
    -- ^ 'ChainDB.getTipBlockNo' for each node at the onset of each slot
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (LedgerUpdate blk)
nodeEventsUpdates     :: ev (LedgerUpdate blk)
    -- ^ Ledger updates every time we adopt a block/switch to a fork
  , forall blk (ev :: * -> *).
NodeEvents blk ev -> ev (TracePipeliningEvent blk)
nodeEventsPipelining  :: ev (ChainDB.TracePipeliningEvent blk)
    -- ^ Pipelining events tracking the tentative header
  }

newNodeInfo ::
  forall blk m.
     IOLike m
  => m ( NodeInfo blk (StrictTMVar m MockFS) (Tracer m)
       , m (NodeInfo blk MockFS [])
       )
newNodeInfo :: forall blk (m :: * -> *).
IOLike m =>
m (NodeInfo blk (StrictTMVar m MockFS) (Tracer m),
   m (NodeInfo blk MockFS []))
newNodeInfo = do
  (nodeInfoEvents, readEvents) <- do
      (t1, m1) <- m (Tracer m (SlotNo, RealPoint blk, BlockNo),
   m [(SlotNo, RealPoint blk, BlockNo)])
forall (m :: * -> *) ev. MonadSTM m => m (Tracer m ev, m [ev])
recordingTracerTVar
      (t2, m2) <- recordingTracerTVar
      (t3, m3) <- recordingTracerTVar
      (t4, m4) <- recordingTracerTVar
      (t5, m5) <- recordingTracerTVar
      (t6, m6) <- recordingTracerTVar
      (t7, m7) <- recordingTracerTVar
      (t8, m8) <- recordingTracerTVar
      pure
          ( NodeEvents     t1     t2     t3     t4     t5     t6     t7     t8
          , NodeEvents <$> m1 <*> m2 <*> m3 <*> m4 <*> m5 <*> m6 <*> m7 <*> m8
          )

  (nodeInfoDBs, readDBs) <- do
      let mk :: m (StrictTMVar m MockFS, STM m MockFS)
          mk = do
              v <- STM m (StrictTMVar m MockFS) -> m (StrictTMVar m MockFS)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (StrictTMVar m MockFS) -> m (StrictTMVar m MockFS))
-> STM m (StrictTMVar m MockFS) -> m (StrictTMVar m MockFS)
forall a b. (a -> b) -> a -> b
$ MockFS -> STM m (StrictTMVar m MockFS)
forall (m :: * -> *) a. MonadSTM m => a -> STM m (StrictTMVar m a)
newTMVar MockFS
Mock.empty
              pure (v, readTMVar v)
      (v1, m1) <- mk
      (v2, m2) <- mk
      (v3, m3) <- mk
      (v4, m4) <- mk
      pure
          ( NodeDBs     v1     v2     v3     v4
          , NodeDBs <$> m1 <*> m2 <*> m3 <*> m4
          )

  pure
      ( NodeInfo{nodeInfoEvents, nodeInfoDBs }
      , NodeInfo <$> readEvents <*> atomically readDBs
      )

{-------------------------------------------------------------------------------
  Test Output - output data about each node
-------------------------------------------------------------------------------}

data NodeOutput blk = NodeOutput
  { forall blk.
NodeOutput blk -> Map SlotNo (Set (RealPoint blk, BlockNo))
nodeOutputAdds         :: Map SlotNo (Set (RealPoint blk, BlockNo))
  , forall blk. NodeOutput blk -> Map SlotNo [CannotForge blk]
nodeOutputCannotForges :: Map SlotNo [CannotForge blk]
  , forall blk. NodeOutput blk -> Chain blk
nodeOutputFinalChain   :: Chain blk
  , forall blk. NodeOutput blk -> LedgerState blk EmptyMK
nodeOutputFinalLedger  :: LedgerState blk EmptyMK
  , forall blk. NodeOutput blk -> Map SlotNo blk
nodeOutputForges       :: Map SlotNo blk
  , forall blk. NodeOutput blk -> Map SlotNo [(RealPoint blk, BlockNo)]
nodeOutputHeaderAdds   :: Map SlotNo [(RealPoint blk, BlockNo)]
  , forall blk.
NodeOutput blk -> Map (RealPoint blk) [ExtValidationError blk]
nodeOutputInvalids     :: Map (RealPoint blk) [ExtValidationError blk]
  , forall blk. NodeOutput blk -> NodeDBs MockFS
nodeOutputNodeDBs      :: NodeDBs MockFS
  , forall blk. NodeOutput blk -> Map SlotNo [(RealPoint blk, BlockNo)]
nodeOutputSelects      :: Map SlotNo [(RealPoint blk, BlockNo)]
  , forall blk. NodeOutput blk -> [LedgerUpdate blk]
nodeOutputUpdates      :: [LedgerUpdate blk]
  , forall blk. NodeOutput blk -> [TracePipeliningEvent blk]
nodePipeliningEvents   :: [ChainDB.TracePipeliningEvent blk]
  }

data TestOutput blk = TestOutput
    { forall blk. TestOutput blk -> Map NodeId (NodeOutput blk)
testOutputNodes       :: Map NodeId (NodeOutput blk)
    , forall blk.
TestOutput blk -> Map SlotNo (Map NodeId (WithOrigin BlockNo))
testOutputTipBlockNos :: Map SlotNo (Map NodeId (WithOrigin BlockNo))
    }

-- | Gather the test output from the nodes
mkTestOutput ::
    forall m blk. (IOLike m, HasHeader blk)
    => [( CoreNodeId
        , m (NodeInfo blk MockFS [])
        , Chain blk
        , LedgerState blk EmptyMK
        )]
    -> m (TestOutput blk)
mkTestOutput :: forall (m :: * -> *) blk.
(IOLike m, HasHeader blk) =>
[(CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
  LedgerState blk EmptyMK)]
-> m (TestOutput blk)
mkTestOutput [(CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
  LedgerState blk EmptyMK)]
vertexInfos = do
    (nodeOutputs', tipBlockNos') <- ([(Map NodeId (NodeOutput blk),
   Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
 -> ([Map NodeId (NodeOutput blk)],
     [Map SlotNo (Map NodeId (WithOrigin BlockNo))]))
-> m [(Map NodeId (NodeOutput blk),
       Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
-> m ([Map NodeId (NodeOutput blk)],
      [Map SlotNo (Map NodeId (WithOrigin BlockNo))])
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap [(Map NodeId (NodeOutput blk),
  Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
-> ([Map NodeId (NodeOutput blk)],
    [Map SlotNo (Map NodeId (WithOrigin BlockNo))])
forall a b. [(a, b)] -> ([a], [b])
unzip (m [(Map NodeId (NodeOutput blk),
     Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
 -> m ([Map NodeId (NodeOutput blk)],
       [Map SlotNo (Map NodeId (WithOrigin BlockNo))]))
-> m [(Map NodeId (NodeOutput blk),
       Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
-> m ([Map NodeId (NodeOutput blk)],
      [Map SlotNo (Map NodeId (WithOrigin BlockNo))])
forall a b. (a -> b) -> a -> b
$ [(CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
  LedgerState blk EmptyMK)]
-> ((CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
     LedgerState blk EmptyMK)
    -> m (Map NodeId (NodeOutput blk),
          Map SlotNo (Map NodeId (WithOrigin BlockNo))))
-> m [(Map NodeId (NodeOutput blk),
       Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM [(CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
  LedgerState blk EmptyMK)]
vertexInfos (((CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
   LedgerState blk EmptyMK)
  -> m (Map NodeId (NodeOutput blk),
        Map SlotNo (Map NodeId (WithOrigin BlockNo))))
 -> m [(Map NodeId (NodeOutput blk),
        Map SlotNo (Map NodeId (WithOrigin BlockNo)))])
-> ((CoreNodeId, m (NodeInfo blk MockFS []), Chain blk,
     LedgerState blk EmptyMK)
    -> m (Map NodeId (NodeOutput blk),
          Map SlotNo (Map NodeId (WithOrigin BlockNo))))
-> m [(Map NodeId (NodeOutput blk),
       Map SlotNo (Map NodeId (WithOrigin BlockNo)))]
forall a b. (a -> b) -> a -> b
$
      \(CoreNodeId
cid, m (NodeInfo blk MockFS [])
readNodeInfo, Chain blk
ch, LedgerState blk EmptyMK
ldgr) -> do
        let nid :: NodeId
nid = CoreNodeId -> NodeId
fromCoreNodeId CoreNodeId
cid
        nodeInfo <- m (NodeInfo blk MockFS [])
readNodeInfo
        let NodeInfo
              { nodeInfoEvents
              , nodeInfoDBs
              } = nodeInfo
        let NodeEvents
              { nodeEventsAdds
              , nodeEventsForges
              , nodeEventsHeaderAdds
              , nodeEventsInvalids
              , nodeEventsSelects
              , nodeEventsTipBlockNos
              , nodeEventsUpdates
              , nodeEventsPipelining
              } = nodeInfoEvents
        let nodeOutput = NodeOutput
              { nodeOutputAdds :: Map SlotNo (Set (RealPoint blk, BlockNo))
nodeOutputAdds        =
                  (Set (RealPoint blk, BlockNo)
 -> Set (RealPoint blk, BlockNo) -> Set (RealPoint blk, BlockNo))
-> [(SlotNo, Set (RealPoint blk, BlockNo))]
-> Map SlotNo (Set (RealPoint blk, BlockNo))
forall k a. Ord k => (a -> a -> a) -> [(k, a)] -> Map k a
Map.fromListWith Set (RealPoint blk, BlockNo)
-> Set (RealPoint blk, BlockNo) -> Set (RealPoint blk, BlockNo)
forall a. Ord a => Set a -> Set a -> Set a
Set.union ([(SlotNo, Set (RealPoint blk, BlockNo))]
 -> Map SlotNo (Set (RealPoint blk, BlockNo)))
-> [(SlotNo, Set (RealPoint blk, BlockNo))]
-> Map SlotNo (Set (RealPoint blk, BlockNo))
forall a b. (a -> b) -> a -> b
$
                  [ (SlotNo
s, (RealPoint blk, BlockNo) -> Set (RealPoint blk, BlockNo)
forall a. a -> Set a
Set.singleton (RealPoint blk
p, BlockNo
bno)) | (SlotNo
s, RealPoint blk
p, BlockNo
bno) <- [(SlotNo, RealPoint blk, BlockNo)]
nodeEventsAdds ]
              , nodeOutputCannotForges :: Map SlotNo [CannotForge blk]
nodeOutputCannotForges =
                  ([CannotForge blk] -> [CannotForge blk] -> [CannotForge blk])
-> [(SlotNo, [CannotForge blk])] -> Map SlotNo [CannotForge blk]
forall k a. Ord k => (a -> a -> a) -> [(k, a)] -> Map k a
Map.fromListWith (([CannotForge blk] -> [CannotForge blk] -> [CannotForge blk])
-> [CannotForge blk] -> [CannotForge blk] -> [CannotForge blk]
forall a b c. (a -> b -> c) -> b -> a -> c
flip [CannotForge blk] -> [CannotForge blk] -> [CannotForge blk]
forall a. [a] -> [a] -> [a]
(++)) ([(SlotNo, [CannotForge blk])] -> Map SlotNo [CannotForge blk])
-> [(SlotNo, [CannotForge blk])] -> Map SlotNo [CannotForge blk]
forall a b. (a -> b) -> a -> b
$
                  [ (SlotNo
s, [CannotForge blk
err]) | TraceNodeCannotForge SlotNo
s CannotForge blk
err <- [TraceForgeEvent blk]
nodeEventsForges ]
              , nodeOutputFinalChain :: Chain blk
nodeOutputFinalChain  = Chain blk
ch
              , nodeOutputFinalLedger :: LedgerState blk EmptyMK
nodeOutputFinalLedger = LedgerState blk EmptyMK
ldgr
              , nodeOutputForges :: Map SlotNo blk
nodeOutputForges      =
                  [(SlotNo, blk)] -> Map SlotNo blk
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([(SlotNo, blk)] -> Map SlotNo blk)
-> [(SlotNo, blk)] -> Map SlotNo blk
forall a b. (a -> b) -> a -> b
$
                  [ (SlotNo
s, blk
b) | TraceForgedBlock SlotNo
s Point blk
_ blk
b MempoolSize
_ <- [TraceForgeEvent blk]
nodeEventsForges ]
              , nodeOutputHeaderAdds :: Map SlotNo [(RealPoint blk, BlockNo)]
nodeOutputHeaderAdds  =
                  ([(RealPoint blk, BlockNo)]
 -> [(RealPoint blk, BlockNo)] -> [(RealPoint blk, BlockNo)])
-> [(SlotNo, [(RealPoint blk, BlockNo)])]
-> Map SlotNo [(RealPoint blk, BlockNo)]
forall k a. Ord k => (a -> a -> a) -> [(k, a)] -> Map k a
Map.fromListWith (([(RealPoint blk, BlockNo)]
 -> [(RealPoint blk, BlockNo)] -> [(RealPoint blk, BlockNo)])
-> [(RealPoint blk, BlockNo)]
-> [(RealPoint blk, BlockNo)]
-> [(RealPoint blk, BlockNo)]
forall a b c. (a -> b -> c) -> b -> a -> c
flip [(RealPoint blk, BlockNo)]
-> [(RealPoint blk, BlockNo)] -> [(RealPoint blk, BlockNo)]
forall a. [a] -> [a] -> [a]
(++)) ([(SlotNo, [(RealPoint blk, BlockNo)])]
 -> Map SlotNo [(RealPoint blk, BlockNo)])
-> [(SlotNo, [(RealPoint blk, BlockNo)])]
-> Map SlotNo [(RealPoint blk, BlockNo)]
forall a b. (a -> b) -> a -> b
$
                  [ (SlotNo
s, [(RealPoint blk
p, BlockNo
bno)])
                  | (SlotNo
s, RealPoint blk
p, BlockNo
bno) <- [(SlotNo, RealPoint blk, BlockNo)]
nodeEventsHeaderAdds
                  ]
              , nodeOutputSelects :: Map SlotNo [(RealPoint blk, BlockNo)]
nodeOutputSelects     =
                  ([(RealPoint blk, BlockNo)]
 -> [(RealPoint blk, BlockNo)] -> [(RealPoint blk, BlockNo)])
-> [(SlotNo, [(RealPoint blk, BlockNo)])]
-> Map SlotNo [(RealPoint blk, BlockNo)]
forall k a. Ord k => (a -> a -> a) -> [(k, a)] -> Map k a
Map.fromListWith (([(RealPoint blk, BlockNo)]
 -> [(RealPoint blk, BlockNo)] -> [(RealPoint blk, BlockNo)])
-> [(RealPoint blk, BlockNo)]
-> [(RealPoint blk, BlockNo)]
-> [(RealPoint blk, BlockNo)]
forall a b c. (a -> b -> c) -> b -> a -> c
flip [(RealPoint blk, BlockNo)]
-> [(RealPoint blk, BlockNo)] -> [(RealPoint blk, BlockNo)]
forall a. [a] -> [a] -> [a]
(++)) ([(SlotNo, [(RealPoint blk, BlockNo)])]
 -> Map SlotNo [(RealPoint blk, BlockNo)])
-> [(SlotNo, [(RealPoint blk, BlockNo)])]
-> Map SlotNo [(RealPoint blk, BlockNo)]
forall a b. (a -> b) -> a -> b
$
                  [ (SlotNo
s, [(RealPoint blk
p, BlockNo
bno)])
                  | (SlotNo
s, RealPoint blk
p, BlockNo
bno) <- [(SlotNo, RealPoint blk, BlockNo)]
nodeEventsSelects
                  ]
              , nodeOutputInvalids :: Map (RealPoint blk) [ExtValidationError blk]
nodeOutputInvalids    = (ExtValidationError blk
-> [ExtValidationError blk] -> [ExtValidationError blk]
forall a. a -> [a] -> [a]
:[]) (ExtValidationError blk -> [ExtValidationError blk])
-> Map (RealPoint blk) (ExtValidationError blk)
-> Map (RealPoint blk) [ExtValidationError blk]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [(RealPoint blk, ExtValidationError blk)]
-> Map (RealPoint blk) (ExtValidationError blk)
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList [(RealPoint blk, ExtValidationError blk)]
nodeEventsInvalids
              , nodeOutputNodeDBs :: NodeDBs MockFS
nodeOutputNodeDBs     = NodeDBs MockFS
nodeInfoDBs
              , nodeOutputUpdates :: [LedgerUpdate blk]
nodeOutputUpdates     = [LedgerUpdate blk]
nodeEventsUpdates
              , nodePipeliningEvents :: [TracePipeliningEvent blk]
nodePipeliningEvents  = [TracePipeliningEvent blk]
nodeEventsPipelining
              }

        pure
          ( Map.singleton nid nodeOutput
          , Map.singleton nid <$> Map.fromList nodeEventsTipBlockNos
          )

    pure $ TestOutput
        { testOutputNodes       = Map.unions nodeOutputs'
        , testOutputTipBlockNos = Map.unionsWith Map.union tipBlockNos'
        }

{-------------------------------------------------------------------------------
  Constraints needed for verbose tracing
-------------------------------------------------------------------------------}

-- | Occurs throughout in positions that might be useful for debugging.
nullDebugTracer :: (Applicative m, Show a) => Tracer m a
nullDebugTracer :: forall (m :: * -> *) a. (Applicative m, Show a) => Tracer m a
nullDebugTracer = Tracer m a
forall (m :: * -> *) a. Applicative m => Tracer m a
nullTracer Tracer m a -> Tracer m a -> Tracer m a
forall a. a -> a -> a
`asTypeOf` Tracer m String -> Tracer m a
forall a (m :: * -> *). Show a => Tracer m String -> Tracer m a
showTracing Tracer m String
forall (m :: * -> *). Applicative m => Tracer m String
debugTracer

-- | Occurs throughout in positions that might be useful for debugging.
nullDebugTracers ::
     ( Monad m
     , Show peer
     , LedgerSupportsProtocol blk
     , TracingConstraints blk
     )
  => Tracers m peer Void blk
nullDebugTracers :: forall (m :: * -> *) peer blk.
(Monad m, Show peer, LedgerSupportsProtocol blk,
 TracingConstraints blk) =>
Tracers m peer Void blk
nullDebugTracers = Tracers m peer Void blk
forall (m :: * -> *) remotePeer localPeer blk.
Monad m =>
Tracers m remotePeer localPeer blk
nullTracers Tracers m peer Void blk
-> Tracers m peer Void blk -> Tracers m peer Void blk
forall a. a -> a -> a
`asTypeOf` Tracer m String -> Tracers m peer Void blk
forall blk remotePeer (m :: * -> *) localPeer.
(Show blk, Show (GenTx blk), Show (Validated (GenTx blk)),
 Show (GenTxId blk), Show (ApplyTxErr blk), Show (Header blk),
 Show (ForgeStateInfo blk), Show (ForgeStateUpdateError blk),
 Show (CannotForge blk), Show remotePeer,
 LedgerSupportsProtocol blk) =>
Tracer m String -> Tracers m remotePeer localPeer blk
showTracers Tracer m String
forall (m :: * -> *). Applicative m => Tracer m String
debugTracer

-- | Occurs throughout in positions that might be useful for debugging.
nullDebugProtocolTracers ::
     ( Monad m
     , HasHeader blk
     , TracingConstraints blk
     , Show peer
     )
  => NTN.Tracers m peer blk failure
nullDebugProtocolTracers :: forall (m :: * -> *) blk peer failure.
(Monad m, HasHeader blk, TracingConstraints blk, Show peer) =>
Tracers m peer blk failure
nullDebugProtocolTracers =
  Tracers m peer blk failure
forall (m :: * -> *) ntnAddr blk e.
Monad m =>
Tracers m ntnAddr blk e
NTN.nullTracers Tracers m peer blk failure
-> Tracers m peer blk failure -> Tracers m peer blk failure
forall a. a -> a -> a
`asTypeOf` Tracer m String -> Tracers m peer blk failure
forall blk ntnAddr (m :: * -> *) e.
(Show blk, Show ntnAddr, Show (Header blk), Show (GenTx blk),
 Show (GenTxId blk), HasHeader blk, HasNestedContent Header blk) =>
Tracer m String -> Tracers m ntnAddr blk e
NTN.showTracers Tracer m String
forall (m :: * -> *). Applicative m => Tracer m String
debugTracer

-- These constraints are when using @showTracer(s) debugTracer@ instead of
-- @nullTracer(s)@.
type TracingConstraints blk =
  ( Show blk
  , Show (ApplyTxErr blk)
  , Show (Header blk)
  , Show (GenTx blk)
  , Show (Validated (GenTx blk))
  , Show (GenTxId blk)
  , Show (ForgeStateInfo blk)
  , Show (ForgeStateUpdateError blk)
  , Show (CannotForge blk)
  , HasNestedContent Header blk
  )

{-------------------------------------------------------------------------------
  Ancillaries
-------------------------------------------------------------------------------}

-- | Spawn multiple async actions and wait for the first one to complete.
--
-- Each child thread is spawned with 'withAsync' and so won't outlive this one.
-- In the use case where each child thread only terminates on an exception, the
-- 'waitAny' ensures that this parent thread will run until a child terminates
-- with an exception, and it will also reraise that exception.
--
-- Why 'NE.NonEmpty'? An empty argument list would have blocked indefinitely,
-- which is likely not intended.
withAsyncsWaitAny :: forall m a. IOLike m => NE.NonEmpty (m a) -> m a
withAsyncsWaitAny :: forall (m :: * -> *) a. IOLike m => NonEmpty (m a) -> m a
withAsyncsWaitAny = [Async m a] -> [m a] -> m a
forall {f :: * -> *} {b}.
MonadAsync f =>
[Async f b] -> [f b] -> f b
go [] ([m a] -> m a)
-> (NonEmpty (m a) -> [m a]) -> NonEmpty (m a) -> m a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NonEmpty (m a) -> [m a]
forall a. NonEmpty a -> [a]
NE.toList
  where
    go :: [Async f b] -> [f b] -> f b
go [Async f b]
acc = \case
      []   -> (Async f b, b) -> b
forall a b. (a, b) -> b
snd ((Async f b, b) -> b) -> f (Async f b, b) -> f b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Async f b] -> f (Async f b, b)
forall a. [Async f a] -> f (Async f a, a)
forall (m :: * -> *) a.
MonadAsync m =>
[Async m a] -> m (Async m a, a)
waitAny [Async f b]
acc
      f b
m:[f b]
ms -> f b -> (Async f b -> f b) -> f b
forall a b. f a -> (Async f a -> f b) -> f b
forall (m :: * -> *) a b.
MonadAsync m =>
m a -> (Async m a -> m b) -> m b
withAsync f b
m ((Async f b -> f b) -> f b) -> (Async f b -> f b) -> f b
forall a b. (a -> b) -> a -> b
$ \Async f b
h -> [Async f b] -> [f b] -> f b
go (Async f b
hAsync f b -> [Async f b] -> [Async f b]
forall a. a -> [a] -> [a]
:[Async f b]
acc) [f b]
ms

-- | The partially instantiation of the 'NetworkApplication' type according to
-- its use in this module
--
-- Used internal to this module, essentially as an abbreviation.
data LimitedApp m addr blk =
   LimitedApp (LimitedApp' m addr blk)

-- | Argument of 'LimitedApp' data constructor
--
-- Used internal to this module, essentially as an abbreviation.
type LimitedApp' m addr blk =
    NTN.Apps m addr
        -- The 'ChainSync' and 'BlockFetch' protocols use @'Serialised' x@ for
        -- the servers and @x@ for the clients. Since both have to match to be
        -- sent across a channel, we can't use @'AnyMessage' ..@, instead, we
        -- (de)serialise the messages so that they can be sent across the
        -- channel with the same type on both ends, i.e., 'Lazy.ByteString'.
        Lazy.ByteString
        Lazy.ByteString
        (AnyMessage (TxSubmission2 (GenTxId blk) (GenTx blk)))
        (AnyMessage KeepAlive)
        (AnyMessage (PeerSharing addr))
        NodeToNodeInitiatorResult ()

{-------------------------------------------------------------------------------
  Tracing
-------------------------------------------------------------------------------}

data MiniProtocolState = MiniProtocolDelayed | MiniProtocolRestarting
  deriving (Int -> MiniProtocolState -> ShowS
[MiniProtocolState] -> ShowS
MiniProtocolState -> String
(Int -> MiniProtocolState -> ShowS)
-> (MiniProtocolState -> String)
-> ([MiniProtocolState] -> ShowS)
-> Show MiniProtocolState
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> MiniProtocolState -> ShowS
showsPrec :: Int -> MiniProtocolState -> ShowS
$cshow :: MiniProtocolState -> String
show :: MiniProtocolState -> String
$cshowList :: [MiniProtocolState] -> ShowS
showList :: [MiniProtocolState] -> ShowS
Show)

-- | Any synchronous exception from a 'directedEdge'
data MiniProtocolFatalException = MiniProtocolFatalException
  { MiniProtocolFatalException -> TypeRep
mpfeType   :: !Typeable.TypeRep
    -- ^ Including the type explicitly makes it easier for a human to debug
  , MiniProtocolFatalException -> SomeException
mpfeExn    :: !SomeException
  , MiniProtocolFatalException -> CoreNodeId
mpfeClient :: !CoreNodeId
  , MiniProtocolFatalException -> CoreNodeId
mpfeServer :: !CoreNodeId
  }
  deriving (Int -> MiniProtocolFatalException -> ShowS
[MiniProtocolFatalException] -> ShowS
MiniProtocolFatalException -> String
(Int -> MiniProtocolFatalException -> ShowS)
-> (MiniProtocolFatalException -> String)
-> ([MiniProtocolFatalException] -> ShowS)
-> Show MiniProtocolFatalException
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> MiniProtocolFatalException -> ShowS
showsPrec :: Int -> MiniProtocolFatalException -> ShowS
$cshow :: MiniProtocolFatalException -> String
show :: MiniProtocolFatalException -> String
$cshowList :: [MiniProtocolFatalException] -> ShowS
showList :: [MiniProtocolFatalException] -> ShowS
Show)

instance Exception MiniProtocolFatalException

-- | Our scheme for Just-In-Time EBBs makes some assumptions
--
data JitEbbError blk
  = JitEbbError (LedgerError blk)
    -- ^ we were unable to extend the ledger state with the JIT EBB

deriving instance LedgerSupportsProtocol blk => Show (JitEbbError blk)
instance LedgerSupportsProtocol blk => Exception (JitEbbError blk)

-- | The 'TxGen' generator consecutively failed too many times
data TxGenFailure = TxGenFailure Int   -- ^ how many times it failed
  deriving (Int -> TxGenFailure -> ShowS
[TxGenFailure] -> ShowS
TxGenFailure -> String
(Int -> TxGenFailure -> ShowS)
-> (TxGenFailure -> String)
-> ([TxGenFailure] -> ShowS)
-> Show TxGenFailure
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> TxGenFailure -> ShowS
showsPrec :: Int -> TxGenFailure -> ShowS
$cshow :: TxGenFailure -> String
show :: TxGenFailure -> String
$cshowList :: [TxGenFailure] -> ShowS
showList :: [TxGenFailure] -> ShowS
Show)

instance Exception TxGenFailure

-- In base@4.20 the Data.List.NonEmpty.unzip is deprecated and suggests that
-- Data.Function.unzip should be used instead,but base versions earlier than
-- 4.20 do not have that.
-- Neatest solution is to cargo cult it here and switch to Data.Function.unzip
-- later.
neUnzip :: Functor f => f (a,b) -> (f a, f b)
neUnzip :: forall (f :: * -> *) a b. Functor f => f (a, b) -> (f a, f b)
neUnzip f (a, b)
xs = ((a, b) -> a
forall a b. (a, b) -> a
fst ((a, b) -> a) -> f (a, b) -> f a
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> f (a, b)
xs, (a, b) -> b
forall a b. (a, b) -> b
snd ((a, b) -> b) -> f (a, b) -> f b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> f (a, b)
xs)