{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE UndecidableInstances #-}

-- | Top-level module to run a single Hydra node.
--
-- Checkout [Hydra
-- Documentation](https://hydra.family/head-protocol/docs/dev/architecture)
-- for some details about the overall architecture of the `Node`.
module Hydra.Node where

import Hydra.Prelude

import Conduit (MonadUnliftIO, ZipSink (..), foldMapC, foldlC, mapC, runConduitRes, sinkList, (.|))
import Control.Concurrent.Class.MonadSTM (
  stateTVar,
  writeTVar,
 )
import Control.Monad.Trans.Writer (execWriter, tell)
import Control.Tracer.JSON (Tracer, traceWith)
import Data.EventSource (EventId, EventSink (..), EventSource (..), getEventId, putEventsToSinks)
import Data.EventSource.Rotation (EventStore (..))
import Data.Secret (mkSecret)
import Data.Text (pack)
import Hydra.API.ClientInput (ClientInput (..))
import Hydra.API.Server (Server, sendMessage)
import Hydra.API.ServerOutput qualified as ServerOutput
import Hydra.Cardano.Api (
  getCardanoPaymentVerificationKey,
 )
import Hydra.Chain (Chain (..), ChainEvent (..), ChainStateHistory (lastKnown), PostTxError, initHistory)
import Hydra.Chain.ChainState (IsChainState (..))
import Hydra.HeadLogic (
  Effect (..),
  HeadState (..),
  Input (..),
  Outcome (..),
  TTL,
  aggregateChainStateHistory,
  aggregateNodeState,
  aggregateState,
 )
import Hydra.HeadLogic qualified as HeadLogic
import Hydra.HeadLogic.Outcome (StateChanged (..), WaitReason (..))
import Hydra.HeadLogic.State (getHeadParameters)
import Hydra.HeadLogic.StateEvent (StateEvent (..))
import Hydra.Ledger (Ledger)
import Hydra.Network (Host (..), Network (..), NetworkCallback (..))
import Hydra.Network qualified as Network
import Hydra.Network.Authenticate (Authenticated (..))
import Hydra.Network.Message (Message (..), NetworkEvent (..))
import Hydra.Node.Environment (Environment (..))
import Hydra.Node.InputQueue (InputQueue (..), Queued (..), createInputQueue)
import Hydra.Node.Outbox (Outbox (..), StallBounds (..), newOutbox)
import Hydra.Node.ParameterMismatch (ParamMismatch (..), ParameterMismatch (..))
import Hydra.Node.State (NodeState (..), initNodeState)
import Hydra.Node.UnsyncedPeriod (UnsyncedPeriod (..))
import Hydra.Node.Util (readFileTextEnvelopeThrow, readSigningKey, readVerificationKey)
import Hydra.Options (CardanoChainConfig (..), ChainConfig (..), RunOptions (..), defaultContestationPeriod, defaultDepositActivation, defaultDepositPeriod)
import Hydra.Tx (HeadParameters (..), Party (..), deriveParty)
import Hydra.Tx.Utils (verificationKeyToOnChainId)

-- * Environment Handling

-- | Initialize the 'Environment' from command line options.
initEnvironment :: RunOptions -> IO Environment
initEnvironment :: RunOptions -> IO Environment
initEnvironment RunOptions
options = do
  -- Wrap the raw key as soon as it leaves disk: every in-process holder
  -- after this point sees only a 'Secret'.
  Secret (SigningKey HydraKey)
sk <- SigningKey HydraKey -> Secret (SigningKey HydraKey)
forall a. a -> Secret a
mkSecret (SigningKey HydraKey -> Secret (SigningKey HydraKey))
-> IO (SigningKey HydraKey) -> IO (Secret (SigningKey HydraKey))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Char] -> IO (SigningKey HydraKey)
forall a. HasTextEnvelope a => [Char] -> IO a
readFileTextEnvelopeThrow [Char]
hydraSigningKey
  [Party]
otherParties <- ([Char] -> IO Party) -> [[Char]] -> IO [Party]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> [a] -> m [b]
mapM [Char] -> IO Party
loadParty [[Char]]
hydraVerificationKeys
  [OnChainId]
participants <- IO [OnChainId]
getParticipants
  Environment -> IO Environment
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Environment -> IO Environment) -> Environment -> IO Environment
forall a b. (a -> b) -> a -> b
$
    Environment
      { $sel:party:Environment :: Party
party = Secret (SigningKey HydraKey) -> Party
deriveParty Secret (SigningKey HydraKey)
sk
      , $sel:signingKey:Environment :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
sk
      , [Party]
otherParties :: [Party]
$sel:otherParties:Environment :: [Party]
otherParties
      , [OnChainId]
participants :: [OnChainId]
$sel:participants:Environment :: [OnChainId]
participants
      , ContestationPeriod
contestationPeriod :: ContestationPeriod
$sel:contestationPeriod:Environment :: ContestationPeriod
contestationPeriod
      , DepositPeriod
depositPeriod :: DepositPeriod
$sel:depositPeriod:Environment :: DepositPeriod
depositPeriod
      , DepositPeriod
depositActivation :: DepositPeriod
$sel:depositActivation:Environment :: DepositPeriod
depositActivation
      , UnsyncedPeriod
unsyncedPeriod :: UnsyncedPeriod
$sel:unsyncedPeriod:Environment :: UnsyncedPeriod
unsyncedPeriod
      , Text
configuredPeers :: Text
$sel:configuredPeers:Environment :: Text
configuredPeers
      }
 where
  -- XXX: This is mostly a cardano-specific initialization step of loading
  -- --cardano-verification-key options and deriving 'OnChainId's from it. We should be able to call out to the various chain layer
  getParticipants :: IO [OnChainId]
getParticipants =
    case ChainConfig
chainConfig of
      Offline{} -> [OnChainId] -> IO [OnChainId]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
      Cardano
        CardanoChainConfig
          { [[Char]]
cardanoVerificationKeys :: [[Char]]
$sel:cardanoVerificationKeys:CardanoChainConfig :: CardanoChainConfig -> [[Char]]
cardanoVerificationKeys
          , [Char]
cardanoSigningKey :: [Char]
$sel:cardanoSigningKey:CardanoChainConfig :: CardanoChainConfig -> [Char]
cardanoSigningKey
          } -> do
          CardanoSigningKey
ownSigningKey <- [Char] -> IO CardanoSigningKey
readSigningKey [Char]
cardanoSigningKey
          [VerificationKey PaymentKey]
otherVerificationKeys <- ([Char] -> IO (VerificationKey PaymentKey))
-> [[Char]] -> IO [VerificationKey PaymentKey]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> [a] -> m [b]
mapM [Char] -> IO (VerificationKey PaymentKey)
readVerificationKey [[Char]]
cardanoVerificationKeys
          [OnChainId] -> IO [OnChainId]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([OnChainId] -> IO [OnChainId]) -> [OnChainId] -> IO [OnChainId]
forall a b. (a -> b) -> a -> b
$ VerificationKey PaymentKey -> OnChainId
verificationKeyToOnChainId (VerificationKey PaymentKey -> OnChainId)
-> [VerificationKey PaymentKey] -> [OnChainId]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (CardanoSigningKey -> VerificationKey PaymentKey
getCardanoPaymentVerificationKey CardanoSigningKey
ownSigningKey VerificationKey PaymentKey
-> [VerificationKey PaymentKey] -> [VerificationKey PaymentKey]
forall a. a -> [a] -> [a]
: [VerificationKey PaymentKey]
otherVerificationKeys)

  contestationPeriod :: ContestationPeriod
contestationPeriod = case ChainConfig
chainConfig of
    Offline{} -> ContestationPeriod
defaultContestationPeriod
    Cardano CardanoChainConfig{$sel:contestationPeriod:CardanoChainConfig :: CardanoChainConfig -> ContestationPeriod
contestationPeriod = ContestationPeriod
cp} -> ContestationPeriod
cp
  depositPeriod :: DepositPeriod
depositPeriod = case ChainConfig
chainConfig of
    Offline{} -> DepositPeriod
defaultDepositPeriod
    Cardano CardanoChainConfig{$sel:depositPeriod:CardanoChainConfig :: CardanoChainConfig -> DepositPeriod
depositPeriod = DepositPeriod
dp} -> DepositPeriod
dp
  depositActivation :: DepositPeriod
depositActivation = case ChainConfig
chainConfig of
    Offline{} -> DepositPeriod
defaultDepositActivation
    Cardano CardanoChainConfig{$sel:depositActivation:CardanoChainConfig :: CardanoChainConfig -> DepositPeriod
depositActivation = DepositPeriod
da} -> DepositPeriod
da
  -- In offline mode, there's no real chain to sync with, so we use a very large
  -- unsynced period to effectively disable the unsynced check.
  unsyncedPeriod :: UnsyncedPeriod
unsyncedPeriod = case ChainConfig
chainConfig of
    Offline{} -> NominalDiffTime -> UnsyncedPeriod
UnsyncedPeriod (Int -> NominalDiffTime
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int
forall a. Bounded a => a
maxBound :: Int))
    Cardano CardanoChainConfig{$sel:unsyncedPeriod:CardanoChainConfig :: CardanoChainConfig -> UnsyncedPeriod
unsyncedPeriod = UnsyncedPeriod
up} -> UnsyncedPeriod
up

  loadParty :: [Char] -> IO Party
loadParty [Char]
p =
    VerificationKey HydraKey -> Party
Party (VerificationKey HydraKey -> Party)
-> IO (VerificationKey HydraKey) -> IO Party
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Char] -> IO (VerificationKey HydraKey)
forall a. HasTextEnvelope a => [Char] -> IO a
readFileTextEnvelopeThrow [Char]
p

  httpUrl :: Host -> [Char]
httpUrl (Host Text
h PortNumber
p) = [Char]
"http://" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Text -> [Char]
forall a. ToString a => a -> [Char]
toString Text
h [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
":" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> PortNumber -> [Char]
forall b a. (Show a, IsString b) => a -> b
show PortNumber
p

  configuredPeers :: Text
configuredPeers =
    [Char] -> Text
pack
      ([Char] -> Text) -> [Char] -> Text
forall a b. (a -> b) -> a -> b
$ [Char] -> [[Char]] -> [Char]
forall a. [a] -> [[a]] -> [a]
intercalate [Char]
","
        ([[Char]] -> [Char]) -> ([Host] -> [[Char]]) -> [Host] -> [Char]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Host -> [Char]) -> [Host] -> [[Char]]
forall a b. (a -> b) -> [a] -> [b]
map (\Host
h -> Host -> [Char]
forall b a. (Show a, IsString b) => a -> b
show Host
h [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
"=" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Host -> [Char]
httpUrl Host
h)
      ([Host] -> [Char]) -> [Host] -> [Char]
forall a b. (a -> b) -> a -> b
$ (Maybe Host -> [Host]
forall a. Maybe a -> [a]
maybeToList Maybe Host
advertise [Host] -> [Host] -> [Host]
forall a. Semigroup a => a -> a -> a
<> [Host]
peers)

  RunOptions
    { [Char]
hydraSigningKey :: [Char]
$sel:hydraSigningKey:RunOptions :: RunOptions -> [Char]
hydraSigningKey
    , [[Char]]
hydraVerificationKeys :: [[Char]]
$sel:hydraVerificationKeys:RunOptions :: RunOptions -> [[Char]]
hydraVerificationKeys
    , ChainConfig
chainConfig :: ChainConfig
$sel:chainConfig:RunOptions :: RunOptions -> ChainConfig
chainConfig
    , Maybe Host
advertise :: Maybe Host
$sel:advertise:RunOptions :: RunOptions -> Maybe Host
advertise
    , [Host]
peers :: [Host]
$sel:peers:RunOptions :: RunOptions -> [Host]
peers
    } = RunOptions
options

-- | Checks that command line options match a given 'HeadState'. This function
-- takes 'Environment' because it is derived from 'RunOptions' via
-- 'initEnvironment'.
--
-- Throws: 'ParameterMismatch' when state not matching the environment.
checkHeadState ::
  MonadThrow m =>
  Tracer m (HydraNodeLog tx) ->
  Environment ->
  HeadState tx ->
  m ()
checkHeadState :: forall (m :: * -> *) tx.
MonadThrow m =>
Tracer m (HydraNodeLog tx) -> Environment -> HeadState tx -> m ()
checkHeadState Tracer m (HydraNodeLog tx)
tracer Environment
env HeadState tx
headState = do
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([ParamMismatch] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [ParamMismatch]
paramsMismatch) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer ([ParamMismatch] -> HydraNodeLog tx
forall tx. [ParamMismatch] -> HydraNodeLog tx
Misconfiguration [ParamMismatch]
paramsMismatch)
    ParameterMismatch -> m ()
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (ParameterMismatch -> m ()) -> ParameterMismatch -> m ()
forall a b. (a -> b) -> a -> b
$ [ParamMismatch] -> ParameterMismatch
ParameterMismatch [ParamMismatch]
paramsMismatch
 where
  paramsMismatch :: [ParamMismatch]
paramsMismatch =
    [ParamMismatch]
-> (HeadParameters -> [ParamMismatch])
-> Maybe HeadParameters
-> [ParamMismatch]
forall b a. b -> (a -> b) -> Maybe a -> b
maybe [] HeadParameters -> [ParamMismatch]
validateParameters (Maybe HeadParameters -> [ParamMismatch])
-> Maybe HeadParameters -> [ParamMismatch]
forall a b. (a -> b) -> a -> b
$ HeadState tx -> Maybe HeadParameters
forall tx. HeadState tx -> Maybe HeadParameters
getHeadParameters HeadState tx
headState

  validateParameters :: HeadParameters -> [ParamMismatch]
validateParameters HeadParameters{$sel:contestationPeriod:HeadParameters :: HeadParameters -> ContestationPeriod
contestationPeriod = ContestationPeriod
loadedCp, $sel:depositPeriod:HeadParameters :: HeadParameters -> DepositPeriod
depositPeriod = DepositPeriod
loadedDp, [Party]
parties :: [Party]
$sel:parties:HeadParameters :: HeadParameters -> [Party]
parties} =
    Writer [ParamMismatch] () -> [ParamMismatch]
forall w a. Writer w a -> w
execWriter (Writer [ParamMismatch] () -> [ParamMismatch])
-> Writer [ParamMismatch] () -> [ParamMismatch]
forall a b. (a -> b) -> a -> b
$ do
      Bool -> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (ContestationPeriod
loadedCp ContestationPeriod -> ContestationPeriod -> Bool
forall a. Eq a => a -> a -> Bool
/= ContestationPeriod
configuredCp) (Writer [ParamMismatch] () -> Writer [ParamMismatch] ())
-> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall a b. (a -> b) -> a -> b
$
        [ParamMismatch] -> Writer [ParamMismatch] ()
forall (m :: * -> *) w. Monad m => w -> WriterT w m ()
tell [ContestationPeriodMismatch{ContestationPeriod
loadedCp :: ContestationPeriod
$sel:loadedCp:ContestationPeriodMismatch :: ContestationPeriod
loadedCp, ContestationPeriod
configuredCp :: ContestationPeriod
$sel:configuredCp:ContestationPeriodMismatch :: ContestationPeriod
configuredCp}]

      Bool -> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (DepositPeriod
loadedDp DepositPeriod -> DepositPeriod -> Bool
forall a. Eq a => a -> a -> Bool
/= DepositPeriod
configuredDp) (Writer [ParamMismatch] () -> Writer [ParamMismatch] ())
-> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall a b. (a -> b) -> a -> b
$
        [ParamMismatch] -> Writer [ParamMismatch] ()
forall (m :: * -> *) w. Monad m => w -> WriterT w m ()
tell [DepositPeriodMismatch{DepositPeriod
loadedDp :: DepositPeriod
$sel:loadedDp:ContestationPeriodMismatch :: DepositPeriod
loadedDp, DepositPeriod
configuredDp :: DepositPeriod
$sel:configuredDp:ContestationPeriodMismatch :: DepositPeriod
configuredDp}]

      let loadedParties :: [Party]
loadedParties = [Party] -> [Party]
forall a. Ord a => [a] -> [a]
sort [Party]
parties
          configuredParties :: [Party]
configuredParties = [Party] -> [Party]
forall a. Ord a => [a] -> [a]
sort (Party
party Party -> [Party] -> [Party]
forall a. a -> [a] -> [a]
: [Party]
otherParties)
      Bool -> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ([Party]
loadedParties [Party] -> [Party] -> Bool
forall a. Eq a => a -> a -> Bool
/= [Party]
configuredParties) (Writer [ParamMismatch] () -> Writer [ParamMismatch] ())
-> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall a b. (a -> b) -> a -> b
$
        [ParamMismatch] -> Writer [ParamMismatch] ()
forall (m :: * -> *) w. Monad m => w -> WriterT w m ()
tell [PartiesMismatch{[Party]
loadedParties :: [Party]
$sel:loadedParties:ContestationPeriodMismatch :: [Party]
loadedParties, [Party]
configuredParties :: [Party]
$sel:configuredParties:ContestationPeriodMismatch :: [Party]
configuredParties}]

  Environment{$sel:contestationPeriod:Environment :: Environment -> ContestationPeriod
contestationPeriod = ContestationPeriod
configuredCp, $sel:depositPeriod:Environment :: Environment -> DepositPeriod
depositPeriod = DepositPeriod
configuredDp, [Party]
$sel:otherParties:Environment :: Environment -> [Party]
otherParties :: [Party]
otherParties, Party
$sel:party:Environment :: Environment -> Party
party :: Party
party} = Environment
env

-- * Create and run a hydra node

-- | A draft version of the 'HydraNode' that holds state, but is not yet
-- connected (see 'connect'). This is commonly created by the 'hydrate' smart
-- constructor.
data DraftHydraNode tx m = DraftHydraNode
  { forall tx (m :: * -> *).
DraftHydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
  , forall tx (m :: * -> *). DraftHydraNode tx m -> Environment
env :: Environment
  , forall tx (m :: * -> *). DraftHydraNode tx m -> Ledger tx
ledger :: Ledger tx
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
  , forall tx (m :: * -> *). DraftHydraNode tx m -> Outbox m
networkOutbox :: Outbox m
  -- ^ Hand-off for 'NetworkEffect's, see 'Outbox'. Created here rather than
  -- in 'connect' so that the API server, which starts first, can report the
  -- live broadcast status in its 'Hydra.API.ServerOutput.Greetings'.
  , -- XXX: This is an odd field in here, but needed for the chain layer to
    -- bootstrap. Maybe move to NodeStateHandler or make it differently accessible?
    forall tx (m :: * -> *).
DraftHydraNode tx m -> ChainStateHistory tx
chainStateHistory :: ChainStateHistory tx
  }

-- | Hydrate a 'DraftHydraNode' by loading events from source, re-aggregate node
-- state and sending events to sinks while doing so.
hydrate ::
  (IsChainState tx, MonadDelay m, MonadLabelledSTM m, MonadAsync m, MonadThrow m, MonadUnliftIO m) =>
  Tracer m (HydraNodeLog tx) ->
  Environment ->
  Ledger tx ->
  ChainStateType tx ->
  EventStore (StateEvent tx) m ->
  [EventSink (StateEvent tx) m] ->
  m (DraftHydraNode tx m)
hydrate :: forall tx (m :: * -> *).
(IsChainState tx, MonadDelay m, MonadLabelledSTM m, MonadAsync m,
 MonadThrow m, MonadUnliftIO m) =>
Tracer m (HydraNodeLog tx)
-> Environment
-> Ledger tx
-> ChainStateType tx
-> EventStore (StateEvent tx) m
-> [EventSink (StateEvent tx) m]
-> m (DraftHydraNode tx m)
hydrate Tracer m (HydraNodeLog tx)
tracer Environment
env Ledger tx
ledger ChainStateType tx
initialChainState EventStore{EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
eventSource :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource, EventSink (StateEvent tx) m
eventSink :: EventSink (StateEvent tx) m
eventSink :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink} [EventSink (StateEvent tx) m]
eventSinks = do
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer HydraNodeLog tx
forall tx. HydraNodeLog tx
LoadingState
  (Last EventId
lastEventId, (NodeState tx
nodeState, ChainStateHistory tx
chainStateHistory)) <-
    ConduitT
  ()
  Void
  (ResourceT m)
  (Last EventId, (NodeState tx, ChainStateHistory tx))
-> m (Last EventId, (NodeState tx, ChainStateHistory tx))
forall (m :: * -> *) r.
MonadUnliftIO m =>
ConduitT () Void (ResourceT m) r -> m r
runConduitRes (ConduitT
   ()
   Void
   (ResourceT m)
   (Last EventId, (NodeState tx, ChainStateHistory tx))
 -> m (Last EventId, (NodeState tx, ChainStateHistory tx)))
-> ConduitT
     ()
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
-> m (Last EventId, (NodeState tx, ChainStateHistory tx))
forall a b. (a -> b) -> a -> b
$
      EventSource (StateEvent tx) m
-> HasEventId (StateEvent tx) =>
   ConduitT () (StateEvent tx) (ResourceT m) ()
forall e (m :: * -> *).
EventSource e m -> HasEventId e => ConduitT () e (ResourceT m) ()
sourceEvents EventSource (StateEvent tx) m
eventSource
        ConduitT () (StateEvent tx) (ResourceT m) ()
-> ConduitT
     (StateEvent tx)
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
-> ConduitT
     ()
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ZipSink
  (StateEvent tx)
  (ResourceT m)
  (Last EventId, (NodeState tx, ChainStateHistory tx))
-> ConduitT
     (StateEvent tx)
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
forall i (m :: * -> *) r. ZipSink i m r -> ConduitT i Void m r
getZipSink
          ( (,)
              (Last EventId
 -> (NodeState tx, ChainStateHistory tx)
 -> (Last EventId, (NodeState tx, ChainStateHistory tx)))
-> ZipSink (StateEvent tx) (ResourceT m) (Last EventId)
-> ZipSink
     (StateEvent tx)
     (ResourceT m)
     ((NodeState tx, ChainStateHistory tx)
      -> (Last EventId, (NodeState tx, ChainStateHistory tx)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ConduitT (StateEvent tx) Void (ResourceT m) (Last EventId)
-> ZipSink (StateEvent tx) (ResourceT m) (Last EventId)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ((StateEvent tx -> Last EventId)
-> ConduitT (StateEvent tx) Void (ResourceT m) (Last EventId)
forall (m :: * -> *) b a o.
(Monad m, Monoid b) =>
(a -> b) -> ConduitT a o m b
foldMapC (Maybe EventId -> Last EventId
forall a. Maybe a -> Last a
Last (Maybe EventId -> Last EventId)
-> (StateEvent tx -> Maybe EventId)
-> StateEvent tx
-> Last EventId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. EventId -> Maybe EventId
forall a. a -> Maybe a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (EventId -> Maybe EventId)
-> (StateEvent tx -> EventId) -> StateEvent tx -> Maybe EventId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StateEvent tx -> EventId
forall a. HasEventId a => a -> EventId
getEventId))
              ZipSink
  (StateEvent tx)
  (ResourceT m)
  ((NodeState tx, ChainStateHistory tx)
   -> (Last EventId, (NodeState tx, ChainStateHistory tx)))
-> ZipSink
     (StateEvent tx) (ResourceT m) (NodeState tx, ChainStateHistory tx)
-> ZipSink
     (StateEvent tx)
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
forall a b.
ZipSink (StateEvent tx) (ResourceT m) (a -> b)
-> ZipSink (StateEvent tx) (ResourceT m) a
-> ZipSink (StateEvent tx) (ResourceT m) b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ConduitT
  (StateEvent tx)
  Void
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
-> ZipSink
     (StateEvent tx) (ResourceT m) (NodeState tx, ChainStateHistory tx)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ConduitT
  (StateEvent tx)
  Void
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
recoverNodeStateC
          )
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ LoadedChainState{$sel:lastKnownChainPoint:BeginInput :: ChainPointType tx
lastKnownChainPoint = ChainStateHistory tx -> ChainPointType tx
forall tx. ChainStateHistory tx -> ChainPointType tx
lastKnown ChainStateHistory tx
chainStateHistory}
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ LoadedState{Last EventId
lastEventId :: Last EventId
$sel:lastEventId:BeginInput :: Last EventId
lastEventId, NodeState tx
nodeState :: NodeState tx
$sel:nodeState:BeginInput :: NodeState tx
nodeState}
  -- Check whether the loaded state matches our configuration (env)
  -- XXX: re-stream events just for this?
  Tracer m (HydraNodeLog tx) -> Environment -> HeadState tx -> m ()
forall (m :: * -> *) tx.
MonadThrow m =>
Tracer m (HydraNodeLog tx) -> Environment -> HeadState tx -> m ()
checkHeadState Tracer m (HydraNodeLog tx)
tracer Environment
env (NodeState tx -> HeadState tx
forall tx. NodeState tx -> HeadState tx
headState NodeState tx
nodeState)
  -- (Re-)submit events to sinks; de-duplication is handled by the sinks
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer HydraNodeLog tx
forall tx. HydraNodeLog tx
ReplayingState
  [StateEvent tx]
replayedEvents <- ConduitT () Void (ResourceT m) [StateEvent tx] -> m [StateEvent tx]
forall (m :: * -> *) r.
MonadUnliftIO m =>
ConduitT () Void (ResourceT m) r -> m r
runConduitRes (ConduitT () Void (ResourceT m) [StateEvent tx]
 -> m [StateEvent tx])
-> ConduitT () Void (ResourceT m) [StateEvent tx]
-> m [StateEvent tx]
forall a b. (a -> b) -> a -> b
$ EventSource (StateEvent tx) m
-> HasEventId (StateEvent tx) =>
   ConduitT () (StateEvent tx) (ResourceT m) ()
forall e (m :: * -> *).
EventSource e m -> HasEventId e => ConduitT () e (ResourceT m) ()
sourceEvents EventSource (StateEvent tx) m
eventSource ConduitT () (StateEvent tx) (ResourceT m) ()
-> ConduitT (StateEvent tx) Void (ResourceT m) [StateEvent tx]
-> ConduitT () Void (ResourceT m) [StateEvent tx]
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ConduitT (StateEvent tx) Void (ResourceT m) [StateEvent tx]
forall (m :: * -> *) a o. Monad m => ConduitT a o m [a]
sinkList
  [EventSink (StateEvent tx) m] -> [StateEvent tx] -> m ()
forall (m :: * -> *) e.
(Monad m, HasEventId e) =>
[EventSink e m] -> [e] -> m ()
putEventsToSinks [EventSink (StateEvent tx) m]
eventSinks [StateEvent tx]
replayedEvents

  NodeStateHandler tx m
nodeStateHandler <- Maybe EventId -> NodeState tx -> m (NodeStateHandler tx m)
forall (m :: * -> *) tx.
MonadLabelledSTM m =>
Maybe EventId -> NodeState tx -> m (NodeStateHandler tx m)
createNodeStateHandler (Last EventId -> Maybe EventId
forall a. Last a -> Maybe a
getLast Last EventId
lastEventId) NodeState tx
nodeState
  InputQueue m (Input tx)
inputQueue <- m (InputQueue m (Input tx))
forall (m :: * -> *) e.
(MonadDelay m, MonadAsync m, MonadLabelledSTM m) =>
m (InputQueue m e)
createInputQueue
  Outbox m
networkOutbox <- StallBounds -> [Char] -> m (Outbox m)
forall (m :: * -> *).
(MonadLabelledSTM m, MonadMonotonicTime m) =>
StallBounds -> [Char] -> m (Outbox m)
newOutbox StallBounds
broadcastStallBounds [Char]
"network-outbox"
  DraftHydraNode tx m -> m (DraftHydraNode tx m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    DraftHydraNode
      { Tracer m (HydraNodeLog tx)
$sel:tracer:DraftHydraNode :: Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer
      , Environment
$sel:env:DraftHydraNode :: Environment
env :: Environment
env
      , Ledger tx
$sel:ledger:DraftHydraNode :: Ledger tx
ledger :: Ledger tx
ledger
      , NodeStateHandler tx m
$sel:nodeStateHandler:DraftHydraNode :: NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
nodeStateHandler
      , InputQueue m (Input tx)
$sel:inputQueue:DraftHydraNode :: InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
inputQueue
      , EventSource (StateEvent tx) m
$sel:eventSource:DraftHydraNode :: EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
eventSource
      , $sel:eventSinks:DraftHydraNode :: [EventSink (StateEvent tx) m]
eventSinks = EventSink (StateEvent tx) m
eventSink EventSink (StateEvent tx) m
-> [EventSink (StateEvent tx) m] -> [EventSink (StateEvent tx) m]
forall a. a -> [a] -> [a]
: [EventSink (StateEvent tx) m]
eventSinks
      , Outbox m
$sel:networkOutbox:DraftHydraNode :: Outbox m
networkOutbox :: Outbox m
networkOutbox
      , ChainStateHistory tx
$sel:chainStateHistory:DraftHydraNode :: ChainStateHistory tx
chainStateHistory :: ChainStateHistory tx
chainStateHistory
      }
 where
  initialState :: NodeState tx
initialState = ChainStateType tx -> NodeState tx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType tx
initialChainState

  recoverNodeStateC :: ConduitT
  (StateEvent tx)
  Void
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
recoverNodeStateC =
    (StateEvent tx -> StateChanged tx)
-> ConduitT (StateEvent tx) (StateChanged tx) (ResourceT m) ()
forall (m :: * -> *) a b. Monad m => (a -> b) -> ConduitT a b m ()
mapC StateEvent tx -> StateChanged tx
forall tx. StateEvent tx -> StateChanged tx
stateChanged
      ConduitT (StateEvent tx) (StateChanged tx) (ResourceT m) ()
-> ConduitT
     (StateChanged tx)
     Void
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
-> ConduitT
     (StateEvent tx)
     Void
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ZipSink
  (StateChanged tx)
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
-> ConduitT
     (StateChanged tx)
     Void
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
forall i (m :: * -> *) r. ZipSink i m r -> ConduitT i Void m r
getZipSink
        ( (,)
            (NodeState tx
 -> ChainStateHistory tx -> (NodeState tx, ChainStateHistory tx))
-> ZipSink (StateChanged tx) (ResourceT m) (NodeState tx)
-> ZipSink
     (StateChanged tx)
     (ResourceT m)
     (ChainStateHistory tx -> (NodeState tx, ChainStateHistory tx))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ConduitT (StateChanged tx) Void (ResourceT m) (NodeState tx)
-> ZipSink (StateChanged tx) (ResourceT m) (NodeState tx)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ((NodeState tx -> StateChanged tx -> NodeState tx)
-> NodeState tx
-> ConduitT (StateChanged tx) Void (ResourceT m) (NodeState tx)
forall (m :: * -> *) a b o.
Monad m =>
(a -> b -> a) -> a -> ConduitT b o m a
foldlC NodeState tx -> StateChanged tx -> NodeState tx
forall tx.
IsChainState tx =>
NodeState tx -> StateChanged tx -> NodeState tx
aggregateNodeState NodeState tx
initialState)
            ZipSink
  (StateChanged tx)
  (ResourceT m)
  (ChainStateHistory tx -> (NodeState tx, ChainStateHistory tx))
-> ZipSink (StateChanged tx) (ResourceT m) (ChainStateHistory tx)
-> ZipSink
     (StateChanged tx)
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
forall a b.
ZipSink (StateChanged tx) (ResourceT m) (a -> b)
-> ZipSink (StateChanged tx) (ResourceT m) a
-> ZipSink (StateChanged tx) (ResourceT m) b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ConduitT
  (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx)
-> ZipSink (StateChanged tx) (ResourceT m) (ChainStateHistory tx)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ((ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx)
-> ChainStateHistory tx
-> ConduitT
     (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx)
forall (m :: * -> *) a b o.
Monad m =>
(a -> b -> a) -> a -> ConduitT b o m a
foldlC ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx
forall tx.
IsChainState tx =>
ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx
aggregateChainStateHistory (ChainStateHistory tx
 -> ConduitT
      (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx))
-> ChainStateHistory tx
-> ConduitT
     (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx)
forall a b. (a -> b) -> a -> b
$ ChainStateType tx -> ChainStateHistory tx
forall tx.
IsChainState tx =>
ChainStateType tx -> ChainStateHistory tx
initHistory ChainStateType tx
initialChainState)
        )

wireChainInput :: DraftHydraNode tx m -> (ChainEvent tx -> m ())
wireChainInput :: forall tx (m :: * -> *).
DraftHydraNode tx m -> ChainEvent tx -> m ()
wireChainInput DraftHydraNode tx m
node = Input tx -> m ()
enqueue (Input tx -> m ())
-> (ChainEvent tx -> Input tx) -> ChainEvent tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChainEvent tx -> Input tx
forall tx. ChainEvent tx -> Input tx
ChainInput
 where
  DraftHydraNode{$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
enqueue :: Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue}} = DraftHydraNode tx m
node

wireClientInput :: DraftHydraNode tx m -> (ClientInput tx -> m ())
wireClientInput :: forall tx (m :: * -> *).
DraftHydraNode tx m -> ClientInput tx -> m ()
wireClientInput DraftHydraNode tx m
node = Input tx -> m ()
enqueue (Input tx -> m ())
-> (ClientInput tx -> Input tx) -> ClientInput tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ClientInput tx -> Input tx
forall tx. ClientInput tx -> Input tx
ClientInput
 where
  DraftHydraNode{$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}} = DraftHydraNode tx m
node

wireNetworkInput :: DraftHydraNode tx m -> NetworkCallback (Authenticated (Message tx)) m
wireNetworkInput :: forall tx (m :: * -> *).
DraftHydraNode tx m
-> NetworkCallback (Authenticated (Message tx)) m
wireNetworkInput DraftHydraNode tx m
node =
  NetworkCallback
    { $sel:deliver:NetworkCallback :: Authenticated (Message tx) -> m ()
deliver = \Authenticated{$sel:party:Authenticated :: forall msg. Authenticated msg -> Party
party = Party
sender, $sel:payload:Authenticated :: forall msg. Authenticated msg -> msg
payload = Message tx
msg} ->
        Input tx -> m ()
enqueue (Input tx -> m ()) -> Input tx -> m ()
forall a b. (a -> b) -> a -> b
$ Party -> Message tx -> Input tx
forall tx. Party -> Message tx -> Input tx
mkNetworkInput Party
sender Message tx
msg
    , $sel:onConnectivity:NetworkCallback :: Connectivity -> m ()
onConnectivity =
        Input tx -> m ()
enqueue (Input tx -> m ())
-> (Connectivity -> Input tx) -> Connectivity -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
1 (NetworkEvent (Message tx) -> Input tx)
-> (Connectivity -> NetworkEvent (Message tx))
-> Connectivity
-> Input tx
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Connectivity -> NetworkEvent (Message tx)
forall msg. Connectivity -> NetworkEvent msg
ConnectivityEvent
    }
 where
  DraftHydraNode{$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}} = DraftHydraNode tx m
node

-- | Create a network input with corresponding default ttl from given sender.
mkNetworkInput :: Party -> Message tx -> Input tx
mkNetworkInput :: forall tx. Party -> Message tx -> Input tx
mkNetworkInput Party
sender Message tx
msg =
  case Message tx
msg of
    ReqTx{} -> TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
defaultTxTTL (NetworkEvent (Message tx) -> Input tx)
-> NetworkEvent (Message tx) -> Input tx
forall a b. (a -> b) -> a -> b
$ ReceivedMessage{Party
sender :: Party
$sel:sender:ConnectivityEvent :: Party
sender, Message tx
msg :: Message tx
$sel:msg:ConnectivityEvent :: Message tx
msg}
    ReqDec{} -> TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
defaultTxTTL (NetworkEvent (Message tx) -> Input tx)
-> NetworkEvent (Message tx) -> Input tx
forall a b. (a -> b) -> a -> b
$ ReceivedMessage{Party
sender :: Party
$sel:sender:ConnectivityEvent :: Party
sender, Message tx
msg :: Message tx
$sel:msg:ConnectivityEvent :: Message tx
msg}
    Message tx
_ -> TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
defaultTTL (NetworkEvent (Message tx) -> Input tx)
-> NetworkEvent (Message tx) -> Input tx
forall a b. (a -> b) -> a -> b
$ ReceivedMessage{Party
sender :: Party
$sel:sender:ConnectivityEvent :: Party
sender, Message tx
msg :: Message tx
$sel:msg:ConnectivityEvent :: Message tx
msg}

-- | Connect chain, network and API to a hydrated 'DraftHydraNode' to get a fully
-- connected 'HydraNode'.
connect ::
  Monad m =>
  Chain tx m ->
  Network m (Message tx) ->
  Server tx m ->
  DraftHydraNode tx m ->
  m (HydraNode tx m)
connect :: forall (m :: * -> *) tx.
Monad m =>
Chain tx m
-> Network m (Message tx)
-> Server tx m
-> DraftHydraNode tx m
-> m (HydraNode tx m)
connect Chain tx m
chain Network m (Message tx)
network Server tx m
server DraftHydraNode tx m
node =
  HydraNode tx m -> m (HydraNode tx m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure HydraNode{Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: Tracer m (HydraNodeLog tx)
tracer, Environment
env :: Environment
$sel:env:HydraNode :: Environment
env, Ledger tx
ledger :: Ledger tx
$sel:ledger:HydraNode :: Ledger tx
ledger, NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
$sel:nodeStateHandler:HydraNode :: NodeStateHandler tx m
nodeStateHandler, InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
$sel:inputQueue:HydraNode :: InputQueue m (Input tx)
inputQueue, EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
$sel:eventSource:HydraNode :: EventSource (StateEvent tx) m
eventSource, [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
$sel:eventSinks:HydraNode :: [EventSink (StateEvent tx) m]
eventSinks, $sel:oc:HydraNode :: Chain tx m
oc = Chain tx m
chain, $sel:hn:HydraNode :: Network m (Message tx)
hn = Network m (Message tx)
network, Server tx m
server :: Server tx m
$sel:server:HydraNode :: Server tx m
server, Outbox m
networkOutbox :: Outbox m
$sel:networkOutbox:HydraNode :: Outbox m
networkOutbox}
 where
  DraftHydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, Environment
$sel:env:DraftHydraNode :: forall tx (m :: * -> *). DraftHydraNode tx m -> Environment
env :: Environment
env, Ledger tx
$sel:ledger:DraftHydraNode :: forall tx (m :: * -> *). DraftHydraNode tx m -> Ledger tx
ledger :: Ledger tx
ledger, NodeStateHandler tx m
$sel:nodeStateHandler:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
nodeStateHandler, InputQueue m (Input tx)
$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
inputQueue, EventSource (StateEvent tx) m
$sel:eventSource:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
eventSource, [EventSink (StateEvent tx) m]
$sel:eventSinks:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
eventSinks, Outbox m
$sel:networkOutbox:DraftHydraNode :: forall tx (m :: * -> *). DraftHydraNode tx m -> Outbox m
networkOutbox :: Outbox m
networkOutbox} = DraftHydraNode tx m
node

-- | Fully connected hydra node with everything wired in.
data HydraNode tx m = HydraNode
  { forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
  , forall tx (m :: * -> *). HydraNode tx m -> Environment
env :: Environment
  , forall tx (m :: * -> *). HydraNode tx m -> Ledger tx
ledger :: Ledger tx
  , forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
  , forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
  , forall tx (m :: * -> *).
HydraNode tx m -> EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
  , forall tx (m :: * -> *).
HydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
  , forall tx (m :: * -> *). HydraNode tx m -> Chain tx m
oc :: Chain tx m
  , forall tx (m :: * -> *). HydraNode tx m -> Network m (Message tx)
hn :: Network m (Message tx)
  , forall tx (m :: * -> *). HydraNode tx m -> Server tx m
server :: Server tx m
  , forall tx (m :: * -> *). HydraNode tx m -> Outbox m
networkOutbox :: Outbox m
  -- ^ Hand-off for 'NetworkEffect's, see 'Outbox'.
  }

-- | Run the network hand-off and its stall monitor concurrently with the
-- given action, stopping when any of them does. An effect throwing therefore
-- still takes the node down, as it did when effects ran on the main loop.
--
-- NOTE: it takes it down asynchronously, though, where running inline threw
-- synchronously from between two effects. The main loop is now cancelled
-- wherever it happens to be, which can be mid-'processStateChanges'. That is
-- the same class of interruption the surrounding 'withChain'/'withNetwork'
-- brackets could already deliver, but this is a new source of it.
withNetworkOutbox :: (MonadAsync m, MonadDelay m) => HydraNode tx m -> m () -> m ()
withNetworkOutbox :: forall (m :: * -> *) tx.
(MonadAsync m, MonadDelay m) =>
HydraNode tx m -> m () -> m ()
withNetworkOutbox HydraNode tx m
node m ()
action =
  ([Char], m ()) -> ([Char], m ()) -> m ()
forall (m :: * -> *) a b.
MonadAsync m =>
([Char], m a) -> ([Char], m b) -> m ()
raceLabelled_
    ([Char]
"network-outbox", Outbox m -> m ()
forall (m :: * -> *). Outbox m -> m ()
runOutbox Outbox m
networkOutbox)
    ([Char]
"hydra-node", ([Char], m ()) -> ([Char], m ()) -> m ()
forall (m :: * -> *) a b.
MonadAsync m =>
([Char], m a) -> ([Char], m b) -> m ()
raceLabelled_ ([Char]
"broadcast-monitor", HydraNode tx m -> m ()
forall (m :: * -> *) tx.
(MonadDelay m, MonadSTM m) =>
HydraNode tx m -> m ()
monitorBroadcast HydraNode tx m
node) ([Char]
"main-loop", m ()
action))
 where
  HydraNode{Outbox m
$sel:networkOutbox:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Outbox m
networkOutbox :: Outbox m
networkOutbox} = HydraNode tx m
node

-- | Report the network hand-off stalling and recovering as 'Connectivity'
-- events, which reach the event log and from there clients already listening.
-- A client connecting mid-stall is told instead by
-- 'Hydra.API.ServerOutput.NetworkInfo', which reads the outbox live rather
-- than replaying these - past outputs are only replayed on request, and a
-- replayed stall may long since have ended.
--
-- These reports travel as network inputs, so while the node is catching up
-- 'updateCatchingUpHead' parks them and a stalled/resumed pair only reaches
-- clients once it is in sync, by which time the stall it describes may be
-- over. 'NetworkInfo' is unaffected, being read live.
--
-- Only a status seen on two consecutive polls is reported, so clients are not
-- flooded (as with the sync status, see #2749). Without that, a network
-- completing something every so often but less often than the stall period
-- flaps between the two reports forever: with completions 15s apart and a 10s
-- poll, the observed gaps cycle 9s, 4s, 14s, giving a report every ~30s
-- indefinitely.
monitorBroadcast :: (MonadDelay m, MonadSTM m) => HydraNode tx m -> m ()
monitorBroadcast :: forall (m :: * -> *) tx.
(MonadDelay m, MonadSTM m) =>
HydraNode tx m -> m ()
monitorBroadcast HydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, Outbox m
$sel:networkOutbox:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Outbox m
networkOutbox :: Outbox m
networkOutbox, $sel:inputQueue:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}} =
  -- Starting from "not stalled" rather than "unknown" keeps a healthy node
  -- silent: there is nothing to correct, because 'Greetings' reads the status
  -- live rather than replaying these.
  Maybe Bool -> Maybe Bool -> Bool -> m ()
go (Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
False) Maybe Bool
forall a. Maybe a
Nothing Bool
False
 where
  -- Polling at the outbox's own stall period bounds how late a report can be
  -- at three times that period, which is ample for something an operator
  -- reads. Floored at a second because 'StallBounds' is caller-supplied and a
  -- zero period would turn this into a busy loop that, under io-sim, stops
  -- virtual time advancing. Below a second the floor is what bounds the
  -- latency, so the "three times" only holds from one second up.
  period :: DiffTime
period = DiffTime -> DiffTime -> DiffTime
forall a. Ord a => a -> a -> a
max DiffTime
1 (DiffTime -> DiffTime)
-> (StallBounds -> DiffTime) -> StallBounds -> DiffTime
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StallBounds -> DiffTime
noProgressFor (StallBounds -> DiffTime) -> StallBounds -> DiffTime
forall a b. (a -> b) -> a -> b
$ Outbox m -> StallBounds
forall (m :: * -> *). Outbox m -> StallBounds
stallBounds Outbox m
networkOutbox

  go :: Maybe Bool -> Maybe Bool -> Bool -> m ()
go Maybe Bool
reported Maybe Bool
lastSeen Bool
hadBacklog = do
    DiffTime -> m ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
period
    Maybe (StallReason, TTL)
stalled <- Outbox m -> m (Maybe (StallReason, TTL))
forall (m :: * -> *). Outbox m -> m (Maybe (StallReason, TTL))
outboxStalled Outbox m
networkOutbox
    Bool
hasBacklog <- Bool -> m Bool
traceBacklog Bool
hadBacklog
    let seen :: Bool
seen = Maybe (StallReason, TTL) -> Bool
forall a. Maybe a -> Bool
isJust Maybe (StallReason, TTL)
stalled
    if Maybe Bool
lastSeen Maybe Bool -> Maybe Bool -> Bool
forall a. Eq a => a -> a -> Bool
== Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
seen Bool -> Bool -> Bool
&& Maybe Bool
reported Maybe Bool -> Maybe Bool -> Bool
forall a. Eq a => a -> a -> Bool
/= Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
seen
      then do
        Connectivity -> m ()
report (Connectivity -> m ()) -> Connectivity -> m ()
forall a b. (a -> b) -> a -> b
$ case Maybe (StallReason, TTL)
stalled of
          Just (StallReason
stallReason, TTL
pendingBroadcasts) -> Network.BroadcastStalled{TTL
pendingBroadcasts :: TTL
$sel:pendingBroadcasts:PeerConnected :: TTL
pendingBroadcasts, StallReason
stallReason :: StallReason
$sel:stallReason:PeerConnected :: StallReason
stallReason}
          Maybe (StallReason, TTL)
Nothing -> Connectivity
Network.BroadcastResumed
        Maybe Bool -> Maybe Bool -> Bool -> m ()
go (Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
seen) (Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
seen) Bool
hasBacklog
      else Maybe Bool -> Maybe Bool -> Bool -> m ()
go Maybe Bool
reported (Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
seen) Bool
hasBacklog

  -- Report what the hand-off holds while it holds anything, and once more when
  -- it empties so the metrics fall back to zero. Silent otherwise: a node
  -- whose network is keeping up would emit this forever for no reader. This
  -- is the fine-grained view the two reports above cannot give, being
  -- transitions of a debounced predicate: it distinguishes a backlog that is
  -- draining slowly from one that is not moving at all, and shows which of
  -- the two 'broadcastStallBounds' limbs is being approached.
  traceBacklog :: Bool -> m Bool
traceBacklog Bool
hadBacklog = do
    (TTL
pendingBroadcasts, DiffTime
noProgressSeconds) <- Outbox m -> m (TTL, DiffTime)
forall (m :: * -> *). Outbox m -> m (TTL, DiffTime)
outboxBacklog Outbox m
networkOutbox
    let hasBacklog :: Bool
hasBacklog = TTL
pendingBroadcasts TTL -> TTL -> Bool
forall a. Ord a => a -> a -> Bool
> TTL
0
    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Bool
hasBacklog Bool -> Bool -> Bool
|| Bool
hadBacklog) (m () -> m ())
-> (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$
      BroadcastBacklog{TTL
pendingBroadcasts :: TTL
$sel:pendingBroadcasts:BeginInput :: TTL
pendingBroadcasts, DiffTime
noProgressSeconds :: DiffTime
$sel:noProgressSeconds:BeginInput :: DiffTime
noProgressSeconds}
    Bool -> m Bool
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
hasBacklog

  -- NOTE: 'enqueue' blocks when the input queue is full, so this loop can
  -- stop polling - most likely when the main loop is itself wedged on the
  -- inline 'postTx', which is exactly when monitoring goes quiet. It cannot
  -- hold up the main loop or the outbox, and both the refusal gate and
  -- 'Greetings' read 'outboxStalled' live rather than through this, so only
  -- the reports and the metrics lag.
  report :: Connectivity -> m ()
report = Input tx -> m ()
enqueue (Input tx -> m ())
-> (Connectivity -> Input tx) -> Connectivity -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
1 (NetworkEvent (Message tx) -> Input tx)
-> (Connectivity -> NetworkEvent (Message tx))
-> Connectivity
-> Input tx
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Connectivity -> NetworkEvent (Message tx)
forall msg. Connectivity -> NetworkEvent msg
ConnectivityEvent

-- | When the node starts refusing the client transactions that grow the
-- outbound backlog, and when it reports the backlog to clients. Ten seconds
-- of no progress, or of backlog at the recent drain rate, is comfortably
-- above the etcd broadcast loop's one second retry and far below any
-- contestation period. The cap on queued messages is the memory backstop:
-- each holds at most a maximum size transaction, so ~160MB at the cap, and it
-- sits well above the bursts a client can fire at a node whose network is
-- keeping up.
--
-- Deliberately not operator-configurable: the useful range is narrow, nothing
-- observable would tell an operator which value to pick, and the natural
-- guess for "off" (zero) is the most aggressive setting rather than the least.
--
-- NOTE: this measures the hand-off, and the shipped 'broadcast' completes as
-- soon as the message is in the network component's own 100-slot
-- pending-broadcast queue. So during an outage the first ~100 messages still
-- complete promptly and nothing is reported; the stall only becomes visible
-- once both queues are saturated, which also puts the real in-flight bound
-- around 'maxPending' plus that queue rather than at 'maxPending'.
broadcastStallBounds :: StallBounds
broadcastStallBounds :: StallBounds
broadcastStallBounds = StallBounds{$sel:noProgressFor:StallBounds :: DiffTime
noProgressFor = DiffTime
10, maxPending :: TTL
maxPending = TTL
10000}

runHydraNode ::
  ( MonadCatch m
  , MonadAsync m
  , MonadDelay m
  , MonadTime m
  , IsChainState tx
  ) =>
  HydraNode tx m ->
  m ()
runHydraNode :: forall (m :: * -> *) tx.
(MonadCatch m, MonadAsync m, MonadDelay m, MonadTime m,
 IsChainState tx) =>
HydraNode tx m -> m ()
runHydraNode node :: HydraNode tx m
node@HydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, Outbox m
$sel:networkOutbox:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Outbox m
networkOutbox :: Outbox m
networkOutbox, $sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler = NodeStateHandler{STM m (NodeState tx)
queryNodeState :: STM m (NodeState tx)
$sel:queryNodeState:NodeStateHandler :: forall tx (m :: * -> *).
NodeStateHandler tx m -> STM m (NodeState tx)
queryNodeState}} = (m () -> m () -> m ()) -> m () -> m () -> m ()
forall a b c. (a -> b -> c) -> b -> a -> c
flip m () -> m () -> m ()
forall a b. m a -> m b -> m a
forall (m :: * -> *) a b. MonadThrow m => m a -> m b -> m a
finally m ()
reportDiscarded (m () -> m ()) -> (m () -> m ()) -> m () -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HydraNode tx m -> m () -> m ()
forall (m :: * -> *) tx.
(MonadAsync m, MonadDelay m) =>
HydraNode tx m -> m () -> m ()
withNetworkOutbox HydraNode tx m
node (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
  -- On startup, resume an interrupted fanout: if the node is mid-fanout
  -- ('FanoutProgress'), re-emit the next fanout step so an auto-drain whose
  -- driver crashed after its last observed chunk continues instead of stalling
  -- (mirrors the rollback re-post). A step already on chain fails harmlessly
  -- ('StalePartialFanoutTx' is silently ignored) and catch-up observations
  -- re-drive; a passive observer ('AwaitingSelection') posts nothing.
  --
  -- Taking only the effects loses nothing: re-posting is not a new decision, so
  -- 'repostFanoutStep' emits no state changes. That matters here because they
  -- could not be applied anyway - 'processStateChanges' writes to the event
  -- sinks, while the in-memory state is updated by 'processNextInput' as it
  -- computes an outcome, and this calls 'repostFanoutStep' directly rather than
  -- going through an input.
  STM m (NodeState tx) -> m (NodeState tx)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m (NodeState tx)
queryNodeState m (NodeState tx) -> (NodeState tx -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \NodeState tx
ns -> case NodeState tx -> HeadState tx
forall tx. NodeState tx -> HeadState tx
headState NodeState tx
ns of
    FanoutProgress PartialFanoutState tx
pfs -> case PartialFanoutState tx -> Outcome tx
forall tx. IsTx tx => PartialFanoutState tx -> Outcome tx
HeadLogic.repostFanoutStep PartialFanoutState tx
pfs of
      Continue{[Effect tx]
effects :: [Effect tx]
$sel:effects:Continue :: forall tx. Outcome tx -> [Effect tx]
effects} -> HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
forall (m :: * -> *) tx.
(MonadAsync m, MonadCatch m, IsChainState tx) =>
HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
processEffects HydraNode tx m
node Tracer m (HydraNodeLog tx)
tracer EventId
0 [Effect tx]
effects
      Outcome tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    HeadState tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  -- NOTE(SN): here we could introduce concurrent head processing, e.g. with
  -- something like 'forM_ [0..1] $ async'
  m () -> m ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    UTCTime
now <- m UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
    UTCTime -> HydraNode tx m -> m ()
forall (m :: * -> *) tx.
(MonadCatch m, MonadAsync m, MonadTime m, IsChainState tx) =>
UTCTime -> HydraNode tx m -> m ()
stepHydraNode UTCTime
now HydraNode tx m
node
 where
  -- Whatever is still queued when we stop is dropped, so say how much: these
  -- messages were accepted from the head logic and never reached the network,
  -- and for a 'ReqSn' or 'AckSn' that leaves a snapshot round announced to
  -- nobody. Usually zero, but not only when the network is refusing work:
  -- 'submit' never blocks, so the main loop routinely runs a few messages
  -- ahead of the drain and stopping right then drops those too. A stall is
  -- what makes the count large.
  reportDiscarded :: m ()
reportDiscarded = do
    TTL
discarded <- STM m TTL -> m TTL
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m TTL -> m TTL) -> STM m TTL -> m TTL
forall a b. (a -> b) -> a -> b
$ Outbox m -> STM m TTL
forall (m :: * -> *). Outbox m -> STM m TTL
pendingActions Outbox m
networkOutbox
    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (TTL
discarded TTL -> TTL -> Bool
forall a. Ord a => a -> a -> Bool
> TTL
0) (m () -> m ())
-> (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ DiscardedBroadcasts{TTL
discarded :: TTL
$sel:discarded:BeginInput :: TTL
discarded}

stepHydraNode ::
  ( MonadCatch m
  , MonadAsync m
  , MonadTime m
  , IsChainState tx
  ) =>
  UTCTime ->
  HydraNode tx m ->
  m ()
stepHydraNode :: forall (m :: * -> *) tx.
(MonadCatch m, MonadAsync m, MonadTime m, IsChainState tx) =>
UTCTime -> HydraNode tx m -> m ()
stepHydraNode UTCTime
now HydraNode tx m
node = do
  i :: Queued (Input tx)
i@Queued{EventId
queuedId :: EventId
$sel:queuedId:Queued :: forall a. Queued a -> EventId
queuedId, Input tx
queuedItem :: Input tx
$sel:queuedItem:Queued :: forall a. Queued a -> a
queuedItem} <- m (Queued (Input tx))
dequeue
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ BeginInput{$sel:by:BeginInput :: Party
by = Party
party, $sel:inputId:BeginInput :: EventId
inputId = EventId
queuedId, $sel:input:BeginInput :: Input tx
input = Input tx
queuedItem}
  Outcome tx
outcome <-
    Input tx -> m (Maybe (Outcome tx))
refuseWhenStalled Input tx
queuedItem m (Maybe (Outcome tx))
-> (Maybe (Outcome tx) -> m (Outcome tx)) -> m (Outcome tx)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Just Outcome tx
refusal -> Outcome tx -> m (Outcome tx)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Outcome tx
refusal
      Maybe (Outcome tx)
Nothing -> STM m (Outcome tx) -> m (Outcome tx)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Outcome tx) -> m (Outcome tx))
-> STM m (Outcome tx) -> m (Outcome tx)
forall a b. (a -> b) -> a -> b
$ HydraNode tx m -> Input tx -> UTCTime -> STM m (Outcome tx)
forall tx (m :: * -> *).
IsChainState tx =>
HydraNode tx m -> Input tx -> UTCTime -> STM m (Outcome tx)
processNextInput HydraNode tx m
node Input tx
queuedItem UTCTime
now
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (Party -> Outcome tx -> HydraNodeLog tx
forall tx. Party -> Outcome tx -> HydraNodeLog tx
LogicOutcome Party
party Outcome tx
outcome)
  case Outcome tx
outcome of
    Continue{[StateChanged tx]
stateChanges :: [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges, [Effect tx]
$sel:effects:Continue :: forall tx. Outcome tx -> [Effect tx]
effects :: [Effect tx]
effects} -> do
      HydraNode tx m -> [StateChanged tx] -> m ()
forall (m :: * -> *) tx.
(MonadSTM m, MonadTime m) =>
HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges HydraNode tx m
node [StateChanged tx]
stateChanges
      HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
forall (m :: * -> *) tx.
(MonadAsync m, MonadCatch m, IsChainState tx) =>
HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
processEffects HydraNode tx m
node Tracer m (HydraNodeLog tx)
tracer EventId
queuedId [Effect tx]
effects
      [StateChanged tx] -> m ()
releaseParkedWhenSynced [StateChanged tx]
stateChanges
    Wait{WaitReason tx
reason :: WaitReason tx
$sel:reason:Continue :: forall tx. Outcome tx -> WaitReason tx
reason, [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges :: [StateChanged tx]
stateChanges} -> do
      HydraNode tx m -> [StateChanged tx] -> m ()
forall (m :: * -> *) tx.
(MonadSTM m, MonadTime m) =>
HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges HydraNode tx m
node [StateChanged tx]
stateChanges
      WaitReason tx -> Queued (Input tx) -> m ()
maybeReenqueue WaitReason tx
reason Queued (Input tx)
i
    Error{} -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer EndInput{$sel:by:BeginInput :: Party
by = Party
party, $sel:inputId:BeginInput :: EventId
inputId = EventId
queuedId}
 where
  -- Refuse the client inputs that grow an outbound backlog we cannot drain,
  -- see 'growsBroadcastBacklog'. Everything else - every chain and network
  -- input, and 'Close', 'Contest' and 'Fanout' - keeps being processed while
  -- the network is down.
  refuseWhenStalled :: Input tx -> m (Maybe (Outcome tx))
refuseWhenStalled = \case
    ClientInput ClientInput tx
clientInput
      | ClientInput tx -> Bool
forall tx. ClientInput tx -> Bool
growsBroadcastBacklog ClientInput tx
clientInput -> do
          -- Only where the input would actually have produced a broadcast.
          -- A node still catching up owes the client its drift, and one whose
          -- head is not open owes it 'CommandFailed' with the state; both are
          -- more use than a retryable "the queue is not draining", and
          -- neither would have broadcast anything.
          --
          -- NOTE: refusing before the head logic runs means a 'Decommit' that
          -- would have been rejected outright (one already in flight, or a tx
          -- the ledger refuses) gets this retryable answer instead of the
          -- definitive 'DecommitInvalid'. Telling them apart needs the very
          -- head-logic pass we are avoiding, and it is only reachable while
          -- stalled.
          Bool
wouldBroadcast <- STM m Bool -> m Bool
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m Bool -> m Bool) -> STM m Bool -> m Bool
forall a b. (a -> b) -> a -> b
$ NodeState tx -> Bool
forall tx. NodeState tx -> Bool
inOpenHeadAndSynced (NodeState tx -> Bool) -> STM m (NodeState tx) -> STM m Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM m (NodeState tx)
queryNodeState
          if Bool -> Bool
not Bool
wouldBroadcast
            then Maybe (Outcome tx) -> m (Maybe (Outcome tx))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Outcome tx)
forall a. Maybe a
Nothing
            else
              Outbox m -> m (Maybe (StallReason, TTL))
forall (m :: * -> *). Outbox m -> m (Maybe (StallReason, TTL))
outboxStalled Outbox m
networkOutbox m (Maybe (StallReason, TTL))
-> (Maybe (StallReason, TTL) -> m (Maybe (Outcome tx)))
-> m (Maybe (Outcome tx))
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
                Maybe (StallReason, TTL)
Nothing -> Maybe (Outcome tx) -> m (Maybe (Outcome tx))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Outcome tx)
forall a. Maybe a
Nothing
                Just (StallReason
stallReason, TTL
pendingBroadcasts) ->
                  Maybe (Outcome tx) -> m (Maybe (Outcome tx))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe (Outcome tx) -> m (Maybe (Outcome tx)))
-> (Outcome tx -> Maybe (Outcome tx))
-> Outcome tx
-> m (Maybe (Outcome tx))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Outcome tx -> Maybe (Outcome tx)
forall a. a -> Maybe a
Just (Outcome tx -> m (Maybe (Outcome tx)))
-> Outcome tx -> m (Maybe (Outcome tx))
forall a b. (a -> b) -> a -> b
$
                    Continue
                      { $sel:stateChanges:Continue :: [StateChanged tx]
stateChanges = []
                      , $sel:effects:Continue :: [Effect tx]
effects = [ClientMessage tx -> Effect tx
forall tx. ClientMessage tx -> Effect tx
ClientEffect ServerOutput.RejectedInputBecauseBroadcastStalled{ClientInput tx
clientInput :: ClientInput tx
$sel:clientInput:CommandFailed :: ClientInput tx
clientInput, StallReason
stallReason :: StallReason
$sel:stallReason:CommandFailed :: StallReason
stallReason, TTL
pendingBroadcasts :: TTL
$sel:pendingBroadcasts:CommandFailed :: TTL
pendingBroadcasts}]
                      }
    Input tx
_ -> Maybe (Outcome tx) -> m (Maybe (Outcome tx))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (Outcome tx)
forall a. Maybe a
Nothing

  maybeReenqueue :: WaitReason tx -> Queued (Input tx) -> m ()
maybeReenqueue WaitReason tx
reason q :: Queued (Input tx)
q@Queued{EventId
$sel:queuedId:Queued :: forall a. Queued a -> EventId
queuedId :: EventId
queuedId, Input tx
$sel:queuedItem:Queued :: forall a. Queued a -> a
queuedItem :: Input tx
queuedItem} =
    case Input tx
queuedItem of
      -- While the node is catching up it can't process any network message
      -- yet, so re-enqueuing here would just spin (and never terminate if the
      -- node stays out of sync). Park the message instead and replay it once
      -- the node is in sync (see 'releaseParkedWhenSynced'). This also avoids
      -- burning the message's retry budget during sync, which would otherwise
      -- drop e.g. ReqTx before catch-up completes.
      NetworkInput TTL
_ NetworkEvent (Message tx)
_
        | WaitOnNodeInSync{} <- WaitReason tx
reason -> Queued (Input tx) -> m ()
park Queued (Input tx)
q
      -- Keep a snapshot request waiting for a settling commit beyond its ttl: it
      -- resolves once this node's chain time catches up, and dropping it would
      -- stall the leader.
      NetworkInput TTL
_ ReceivedMessage{$sel:msg:ConnectivityEvent :: forall msg. NetworkEvent msg -> msg
msg = ReqSn{}}
        | WaitOnUnresolvedCommit{} <- WaitReason tx
reason -> DiffTime -> Queued (Input tx) -> m ()
reenqueue DiffTime
waitDelay Queued (Input tx)
q
      NetworkInput TTL
ttl NetworkEvent (Message tx)
msg
        | TTL
ttl TTL -> TTL -> Bool
forall a. Ord a => a -> a -> Bool
> TTL
0 -> DiffTime -> Queued (Input tx) -> m ()
reenqueue DiffTime
waitDelay Queued (Input tx)
q{queuedItem = NetworkInput (ttl - 1) msg}
      Input tx
_ -> Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ DroppedFromQueue{$sel:inputId:BeginInput :: EventId
inputId = EventId
queuedId, $sel:input:BeginInput :: Input tx
input = Input tx
queuedItem}

  -- Replay parked network inputs once the node transitions into sync, so
  -- messages received during catch-up are processed rather than lost.
  releaseParkedWhenSynced :: [StateChanged tx] -> m ()
releaseParkedWhenSynced [StateChanged tx]
stateChanges =
    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ((StateChanged tx -> Bool) -> [StateChanged tx] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any StateChanged tx -> Bool
forall tx. StateChanged tx -> Bool
isNodeSynced [StateChanged tx]
stateChanges) m ()
releaseParked

  isNodeSynced :: StateChanged tx -> Bool
  isNodeSynced :: forall tx. StateChanged tx -> Bool
isNodeSynced = \case
    NodeSynced{} -> Bool
True
    StateChanged tx
_ -> Bool
False

  Environment{Party
$sel:party:Environment :: Environment -> Party
party :: Party
party} = Environment
env

  HydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, $sel:inputQueue:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{m (Queued (Input tx))
dequeue :: m (Queued (Input tx))
$sel:dequeue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> m (Queued e)
dequeue, DiffTime -> Queued (Input tx) -> m ()
reenqueue :: DiffTime -> Queued (Input tx) -> m ()
$sel:reenqueue:InputQueue :: forall (m :: * -> *) e.
InputQueue m e -> DiffTime -> Queued e -> m ()
reenqueue, Queued (Input tx) -> m ()
park :: Queued (Input tx) -> m ()
$sel:park:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> Queued e -> m ()
park, m ()
releaseParked :: m ()
$sel:releaseParked:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> m ()
releaseParked}, Environment
$sel:env:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Environment
env :: Environment
env, Outbox m
$sel:networkOutbox:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Outbox m
networkOutbox :: Outbox m
networkOutbox, $sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler = NodeStateHandler{STM m (NodeState tx)
$sel:queryNodeState:NodeStateHandler :: forall tx (m :: * -> *).
NodeStateHandler tx m -> STM m (NodeState tx)
queryNodeState :: STM m (NodeState tx)
queryNodeState}} = HydraNode tx m
node

-- | Client inputs that turn into a broadcast directly, and so are the ones
-- worth refusing: the protocol's own messages cannot pile up while the
-- network is down, because 'broadcast' is self-delivering, so our own 'AckSn'
-- never comes back, the confirmed snapshot number freezes and
-- 'snapshotInFlight' caps both 'ReqSn' and 'AckSn' at one apiece.
--
-- NOTE: that cap is not airtight, and this is not a complete bound.
-- 'onOpenChainTick' emits a 'ReqSn' on a chain tick with no network input at
-- all, and 'SideLoadSnapshot' - which broadcasts nothing itself, so is not
-- listed here - clears exactly the state that cap reads. A client looping
-- 'SideLoadSnapshot' can therefore draw one further 'ReqSn' per tick. Left
-- ungated on purpose: side-loading is the documented recovery for a 'ReqSn'
-- or 'AckSn' lost from the hand-off, so refusing it while stalled would block
-- the way out. The leak is one small message per chain tick.
growsBroadcastBacklog :: ClientInput tx -> Bool
growsBroadcastBacklog :: forall tx. ClientInput tx -> Bool
growsBroadcastBacklog = \case
  NewTx{} -> Bool
True
  Decommit{} -> Bool
True
  ClientInput tx
_ -> Bool
False

-- | Whether 'growsBroadcastBacklog' inputs would reach the code that
-- broadcasts, rather than being answered by the sync or head-state checks.
inOpenHeadAndSynced :: NodeState tx -> Bool
inOpenHeadAndSynced :: forall tx. NodeState tx -> Bool
inOpenHeadAndSynced = \case
  NodeInSync{$sel:headState:NodeInSync :: forall tx. NodeState tx -> HeadState tx
headState = Open{}} -> Bool
True
  NodeState tx
_ -> Bool
False

-- | The maximum number of times to re-enqueue a network messages upon 'Wait'.
-- outcome.
defaultTTL :: TTL
defaultTTL :: TTL
defaultTTL = TTL
6000

-- | The maximum number of times to re-enqueue 'ReqTx' and 'ReqDec' network
-- messages upon 'Wait'.
defaultTxTTL :: TTL
defaultTxTTL :: TTL
defaultTxTTL = TTL
5

-- | The time to wait between re-enqueuing a 'Wait' outcome.
waitDelay :: DiffTime
waitDelay :: DiffTime
waitDelay = DiffTime
0.1

-- | Monadic interface around 'Hydra.Logic.update'.
processNextInput ::
  IsChainState tx =>
  HydraNode tx m ->
  Input tx ->
  UTCTime ->
  STM m (Outcome tx)
processNextInput :: forall tx (m :: * -> *).
IsChainState tx =>
HydraNode tx m -> Input tx -> UTCTime -> STM m (Outcome tx)
processNextInput HydraNode{NodeStateHandler tx m
$sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
nodeStateHandler, Ledger tx
$sel:ledger:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Ledger tx
ledger :: Ledger tx
ledger, Environment
$sel:env:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Environment
env :: Environment
env} Input tx
e UTCTime
now =
  (NodeState tx -> (Outcome tx, NodeState tx)) -> STM m (Outcome tx)
forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState ((NodeState tx -> (Outcome tx, NodeState tx))
 -> STM m (Outcome tx))
-> (NodeState tx -> (Outcome tx, NodeState tx))
-> STM m (Outcome tx)
forall a b. (a -> b) -> a -> b
$ \NodeState tx
s ->
    let outcome :: Outcome tx
outcome = Environment
-> Ledger tx -> UTCTime -> NodeState tx -> Input tx -> Outcome tx
forall tx.
IsChainState tx =>
Environment
-> Ledger tx -> UTCTime -> NodeState tx -> Input tx -> Outcome tx
HeadLogic.update Environment
env Ledger tx
ledger UTCTime
now NodeState tx
s Input tx
e
     in (Outcome tx
outcome, NodeState tx -> Outcome tx -> NodeState tx
forall tx.
IsChainState tx =>
NodeState tx -> Outcome tx -> NodeState tx
aggregateState NodeState tx
s Outcome tx
outcome)
 where
  NodeStateHandler{forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState :: forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
$sel:modifyNodeState:NodeStateHandler :: forall tx (m :: * -> *).
NodeStateHandler tx m
-> forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState} = NodeStateHandler tx m
nodeStateHandler

processStateChanges :: (MonadSTM m, MonadTime m) => HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges :: forall (m :: * -> *) tx.
(MonadSTM m, MonadTime m) =>
HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges HydraNode tx m
node [StateChanged tx]
stateChanges = do
  [StateEvent tx]
events <- [StateChanged tx]
-> (StateChanged tx -> m (StateEvent tx)) -> m [StateEvent tx]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM [StateChanged tx]
stateChanges ((StateChanged tx -> m (StateEvent tx)) -> m [StateEvent tx])
-> (StateChanged tx -> m (StateEvent tx)) -> m [StateEvent tx]
forall a b. (a -> b) -> a -> b
$ \StateChanged tx
stateChanged -> do
    UTCTime
time <- m UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
    EventId
eventId <- STM m EventId -> m EventId
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m EventId
getNextEventId
    StateEvent tx -> m (StateEvent tx)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure StateEvent{EventId
eventId :: EventId
$sel:eventId:StateEvent :: EventId
eventId, StateChanged tx
$sel:stateChanged:StateEvent :: StateChanged tx
stateChanged :: StateChanged tx
stateChanged, UTCTime
time :: UTCTime
$sel:time:StateEvent :: UTCTime
time}
  [EventSink (StateEvent tx) m] -> [StateEvent tx] -> m ()
forall (m :: * -> *) e.
(Monad m, HasEventId e) =>
[EventSink e m] -> [e] -> m ()
putEventsToSinks [EventSink (StateEvent tx) m]
eventSinks [StateEvent tx]
events
 where
  HydraNode
    { [EventSink (StateEvent tx) m]
$sel:eventSinks:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
eventSinks
    , $sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler = NodeStateHandler{STM m EventId
getNextEventId :: STM m EventId
$sel:getNextEventId:NodeStateHandler :: forall tx (m :: * -> *). NodeStateHandler tx m -> STM m EventId
getNextEventId}
    } = HydraNode tx m
node

processEffects ::
  ( MonadAsync m
  , MonadCatch m
  , IsChainState tx
  ) =>
  HydraNode tx m ->
  Tracer m (HydraNodeLog tx) ->
  Word64 ->
  [Effect tx] ->
  m ()
processEffects :: forall (m :: * -> *) tx.
(MonadAsync m, MonadCatch m, IsChainState tx) =>
HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
processEffects HydraNode tx m
node Tracer m (HydraNodeLog tx)
tracer EventId
inputId [Effect tx]
effects = do
  ((Effect tx, Word32) -> m ()) -> [(Effect tx, Word32)] -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (Effect tx, Word32) -> m ()
processEffect ([(Effect tx, Word32)] -> m ()) -> [(Effect tx, Word32)] -> m ()
forall a b. (a -> b) -> a -> b
$ [Effect tx] -> [Word32] -> [(Effect tx, Word32)]
forall a b. [a] -> [b] -> [(a, b)]
zip [Effect tx]
effects [Word32
0 ..]
 where
  processEffect :: (Effect tx, Word32) -> m ()
processEffect (Effect tx
effect, Word32
effectId) = do
    Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ Party -> EventId -> Word32 -> Effect tx -> HydraNodeLog tx
forall tx.
Party -> EventId -> Word32 -> Effect tx -> HydraNodeLog tx
BeginEffect Party
party EventId
inputId Word32
effectId Effect tx
effect
    case Effect tx
effect of
      -- NOTE: 'sendMessage' is a write to an unbounded channel, so it cannot
      -- block. It also shares that channel with the server outputs this loop
      -- emits through the event sinks, so handing it off would reorder the
      -- two as clients see them.
      ClientEffect ClientMessage tx
i -> Server tx m -> ClientMessage tx -> m ()
forall tx (m :: * -> *). Server tx m -> ClientMessage tx -> m ()
sendMessage Server tx m
server ClientMessage tx
i
      -- NOTE: from here on 'EndEffect' means the effect was accepted for
      -- execution, not that it completed.
      NetworkEffect Message tx
msg -> Outbox m -> m () -> m ()
forall (m :: * -> *). Outbox m -> m () -> m ()
submit Outbox m
networkOutbox (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ Network m (Message tx) -> Message tx -> m ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network m (Message tx)
hn Message tx
msg
      -- NOTE: still inline, and 'postTx' can block - the direct backend waits
      -- on the tx-submission response. So a stalled chain backend still stops
      -- this loop, which is the sibling advisory on the L1 submission queue,
      -- not this one. Handing it off does not fix that either: 'Close' and
      -- 'Contest' are 'OnChainEffect's too, so they would queue behind the
      -- same stall, and the transaction would then be built against a later
      -- chain state than the input was decided on.
      OnChainEffect{PostChainTx tx
postChainTx :: PostChainTx tx
$sel:postChainTx:ClientEffect :: forall tx. Effect tx -> PostChainTx tx
postChainTx} ->
        MonadThrow m => PostChainTx tx -> m ()
PostChainTx tx -> m ()
postTx PostChainTx tx
postChainTx
          m () -> (PostTxError tx -> m ()) -> m ()
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` \(PostTxError tx
postTxError :: PostTxError tx) ->
            Input tx -> m ()
enqueue (Input tx -> m ())
-> (ChainEvent tx -> Input tx) -> ChainEvent tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChainEvent tx -> Input tx
forall tx. ChainEvent tx -> Input tx
ChainInput (ChainEvent tx -> m ()) -> ChainEvent tx -> m ()
forall a b. (a -> b) -> a -> b
$ PostTxError{PostChainTx tx
postChainTx :: PostChainTx tx
$sel:postChainTx:Observation :: PostChainTx tx
postChainTx, PostTxError tx
postTxError :: PostTxError tx
$sel:postTxError:Observation :: PostTxError tx
postTxError, $sel:failingTx:Observation :: Maybe tx
failingTx = Maybe tx
forall a. Maybe a
Nothing}
    Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ Party -> EventId -> Word32 -> HydraNodeLog tx
forall tx. Party -> EventId -> Word32 -> HydraNodeLog tx
EndEffect Party
party EventId
inputId Word32
effectId

  HydraNode
    { Network m (Message tx)
$sel:hn:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Network m (Message tx)
hn :: Network m (Message tx)
hn
    , $sel:oc:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Chain tx m
oc = Chain{MonadThrow m => PostChainTx tx -> m ()
postTx :: MonadThrow m => PostChainTx tx -> m ()
$sel:postTx:Chain :: forall tx (m :: * -> *).
Chain tx m -> MonadThrow m => PostChainTx tx -> m ()
postTx}
    , $sel:inputQueue:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}
    , $sel:env:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Environment
env = Environment{Party
$sel:party:Environment :: Environment -> Party
party :: Party
party}
    , Server tx m
$sel:server:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Server tx m
server :: Server tx m
server
    , Outbox m
$sel:networkOutbox:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Outbox m
networkOutbox :: Outbox m
networkOutbox
    } = HydraNode tx m
node

-- ** Manage state

-- | Handle to access and modify the state in the Hydra Node.
data NodeStateHandler tx m = NodeStateHandler
  { forall tx (m :: * -> *).
NodeStateHandler tx m
-> forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState :: forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
  , forall tx (m :: * -> *).
NodeStateHandler tx m -> STM m (NodeState tx)
queryNodeState :: STM m (NodeState tx)
  , forall tx (m :: * -> *). NodeStateHandler tx m -> STM m EventId
getNextEventId :: STM m EventId
  }

-- | Initialize a new 'NodeStateHandler'.
createNodeStateHandler ::
  MonadLabelledSTM m =>
  -- | Last seen 'EventId'.
  Maybe EventId ->
  NodeState tx ->
  m (NodeStateHandler tx m)
createNodeStateHandler :: forall (m :: * -> *) tx.
MonadLabelledSTM m =>
Maybe EventId -> NodeState tx -> m (NodeStateHandler tx m)
createNodeStateHandler Maybe EventId
lastSeenEventId NodeState tx
initialState = do
  TVar m EventId
nextEventIdV <- [Char] -> EventId -> m (TVar m EventId)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
[Char] -> a -> m (TVar m a)
newLabelledTVarIO [Char]
"next-event-id" (EventId -> m (TVar m EventId)) -> EventId -> m (TVar m EventId)
forall a b. (a -> b) -> a -> b
$ EventId -> (EventId -> EventId) -> Maybe EventId -> EventId
forall b a. b -> (a -> b) -> Maybe a -> b
maybe EventId
0 (EventId -> EventId -> EventId
forall a. Num a => a -> a -> a
+ EventId
1) Maybe EventId
lastSeenEventId
  TVar m (NodeState tx)
ns <- [Char] -> NodeState tx -> m (TVar m (NodeState tx))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
[Char] -> a -> m (TVar m a)
newLabelledTVarIO [Char]
"node-state" NodeState tx
initialState
  NodeStateHandler tx m -> m (NodeStateHandler tx m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    NodeStateHandler
      { $sel:modifyNodeState:NodeStateHandler :: forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState = TVar m (NodeState tx)
-> (NodeState tx -> (a, NodeState tx)) -> STM m a
forall s a. TVar m s -> (s -> (a, s)) -> STM m a
forall (m :: * -> *) s a.
MonadSTM m =>
TVar m s -> (s -> (a, s)) -> STM m a
stateTVar TVar m (NodeState tx)
ns
      , $sel:queryNodeState:NodeStateHandler :: STM m (NodeState tx)
queryNodeState = TVar m (NodeState tx) -> STM m (NodeState tx)
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m (NodeState tx)
ns
      , $sel:getNextEventId:NodeStateHandler :: STM m EventId
getNextEventId = do
          EventId
eventId <- TVar m EventId -> STM m EventId
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m EventId
nextEventIdV
          TVar m EventId -> EventId -> STM m ()
forall a. TVar m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar m EventId
nextEventIdV (EventId -> STM m ()) -> EventId -> STM m ()
forall a b. (a -> b) -> a -> b
$ EventId
eventId EventId -> EventId -> EventId
forall a. Num a => a -> a -> a
+ EventId
1
          EventId -> STM m EventId
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure EventId
eventId
      }

-- * Logging

data HydraNodeLog tx
  = BeginInput {forall tx. HydraNodeLog tx -> Party
by :: Party, forall tx. HydraNodeLog tx -> EventId
inputId :: Word64, forall tx. HydraNodeLog tx -> Input tx
input :: Input tx}
  | EndInput {by :: Party, inputId :: Word64}
  | BeginEffect {by :: Party, inputId :: Word64, forall tx. HydraNodeLog tx -> Word32
effectId :: Word32, forall tx. HydraNodeLog tx -> Effect tx
effect :: Effect tx}
  | EndEffect {by :: Party, inputId :: Word64, effectId :: Word32}
  | LogicOutcome {by :: Party, forall tx. HydraNodeLog tx -> Outcome tx
outcome :: Outcome tx}
  | DroppedFromQueue {inputId :: Word64, input :: Input tx}
  | LoadingState
  | LoadedState {forall tx. HydraNodeLog tx -> Last EventId
lastEventId :: Last EventId, forall tx. HydraNodeLog tx -> NodeState tx
nodeState :: NodeState tx}
  | LoadedChainState {forall tx. HydraNodeLog tx -> ChainPointType tx
lastKnownChainPoint :: ChainPointType tx}
  | ReplayingState
  | Misconfiguration {forall tx. HydraNodeLog tx -> [ParamMismatch]
misconfigurationErrors :: [ParamMismatch]}
  | -- | Outbound messages accepted from the head logic but never handed to
    -- the network, dropped because the node is stopping.
    DiscardedBroadcasts {forall tx. HydraNodeLog tx -> TTL
discarded :: Natural}
  | -- | How much the outbound hand-off is holding, and for how long it has
    -- completed nothing. Emitted by 'monitorBroadcast' only while there is a
    -- backlog, plus once when one clears, so a node whose network is fine
    -- says nothing. Drives the broadcast metrics; see 'Hydra.Logging.Monitoring'.
    BroadcastBacklog {forall tx. HydraNodeLog tx -> TTL
pendingBroadcasts :: Natural, forall tx. HydraNodeLog tx -> DiffTime
noProgressSeconds :: DiffTime}
  deriving stock ((forall x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x)
-> (forall x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx)
-> Generic (HydraNodeLog tx)
forall x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
forall x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
forall tx x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
forall tx x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
$cfrom :: forall tx x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
from :: forall x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
$cto :: forall tx x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
to :: forall x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
Generic)

deriving stock instance IsChainState tx => Eq (HydraNodeLog tx)
deriving stock instance IsChainState tx => Show (HydraNodeLog tx)
deriving anyclass instance IsChainState tx => ToJSON (HydraNodeLog tx)