Safe Haskell	Safe-Inferred
Language	GHC2021

Hydra.Network.Heartbeat

Description

An implementation of an application-level failure detector. This module exposes a Component withHeartbeat than can be used to wrap another NetworkComponent and piggy-back on it to send and propagate Heartbeat messages and detect other parties' liveness.

It is inspired by the Increasing timeout algorithms from the book Introduction to Reliable and Secure Distributed Programming by Cachin et al. which is an Eventually Perfect Failure Detector suitable for partially synchronous network settings. It has the following behaviour:

It broadcasts a Ping to other parties through the underlying Network implementation if the last message has been sent more than 3s ago
When receiving messages from other parties, it records reception time and notifies underlying node with a Connected message
If new messages are received from $sel:alive:HeartbeatState parties before 3s timeout expires no new Connected message is sent
If main thread detects that a formerly $sel:alive:HeartbeatState party has not been seen for more than 3s, it is marked as $sel:suspected:HeartbeatState and a Disconnected message is sent to the node.

Synopsis

data HeartbeatState = HeartbeatState {
- alive :: Map NodeId Time
- suspected :: Set NodeId
}
initialHeartbeatState :: HeartbeatState
data Heartbeat msg
- = Data NodeId msg
- | Ping NodeId
isPing :: Heartbeat msg -> Bool
heartbeatDelay :: DiffTime
livenessDelay :: DiffTime
withHeartbeat :: (MonadAsync m, MonadDelay m) => NodeId -> NetworkComponent m (Heartbeat inbound) (Heartbeat outbound) a -> NetworkComponent m (Either Connectivity inbound) outbound a
updateStateFromIncomingMessages :: (MonadSTM m, MonadMonotonicTime m) => TVar m HeartbeatState -> NetworkCallback (Either Connectivity inbound) m -> NetworkCallback (Heartbeat inbound) m
updateStateFromOutgoingMessages :: (MonadSTM m, MonadMonotonicTime m) => NodeId -> TVar m (Maybe Time) -> Network m (Heartbeat outbound) -> Network m outbound
updateLastSent :: MonadSTM m => TVar m (Maybe Time) -> Time -> m ()
checkHeartbeatState :: (MonadDelay m, MonadSTM m) => NodeId -> TVar m (Maybe Time) -> Network m (Heartbeat outbound) -> m ()
shouldSendHeartbeat :: Time -> Maybe Time -> Bool
checkRemoteParties :: (MonadDelay m, MonadSTM m) => TVar m HeartbeatState -> (Connectivity -> m ()) -> m ()
updateSuspected :: MonadSTM m => TVar m HeartbeatState -> Time -> m (Set NodeId)
checkTimeout :: (DiffTime -> DiffTime) -> Time -> Time -> Bool

Documentation

data HeartbeatState Source #

Constructors

HeartbeatState
Fields alive :: Map NodeId Time The map of known `Connected` parties with the last time they've been "seen". This is updated when we see a message from another node suspected :: Set NodeId The set of known parties which might be `Disconnected` This is updated after some time no message has been received from a node.

Instances

Instances details

Eq HeartbeatState Source #
Instance details Defined in Hydra.Network.Heartbeat Methods (==) :: HeartbeatState -> HeartbeatState -> Bool Source # (/=) :: HeartbeatState -> HeartbeatState -> Bool Source #

initialHeartbeatState :: HeartbeatState Source #

data Heartbeat msg Source #

Constructors

Data NodeId msg
Ping NodeId

Instances

Instances details

FromJSON msg => FromJSON (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods parseJSON :: Value -> Parser (Heartbeat msg) parseJSONList :: Value -> Parser [Heartbeat msg] omittedField :: Maybe (Heartbeat msg)
ToJSON msg => ToJSON (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods toJSON :: Heartbeat msg -> Value toEncoding :: Heartbeat msg -> Encoding toJSONList :: [Heartbeat msg] -> Value toEncodingList :: [Heartbeat msg] -> Encoding omitField :: Heartbeat msg -> Bool
Generic (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Associated Types type Rep (Heartbeat msg) :: Type -> Type Source # Methods from :: Heartbeat msg -> Rep (Heartbeat msg) x Source # to :: Rep (Heartbeat msg) x -> Heartbeat msg Source #
Show msg => Show (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods showsPrec :: Int -> Heartbeat msg -> ShowS Source # show :: Heartbeat msg -> String Source # showList :: [Heartbeat msg] -> ShowS Source #
FromCBOR msg => FromCBOR (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods fromCBOR :: Decoder s (Heartbeat msg) label :: Proxy (Heartbeat msg) -> Text
ToCBOR msg => ToCBOR (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods toCBOR :: Heartbeat msg -> Encoding encodedSizeExpr :: (forall t. ToCBOR t => Proxy t -> Size) -> Proxy (Heartbeat msg) -> Size encodedListSizeExpr :: (forall t. ToCBOR t => Proxy t -> Size) -> Proxy [Heartbeat msg] -> Size
ToCBOR msg => SignableRepresentation (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods getSignableRepresentation :: Heartbeat msg -> ByteString
Eq msg => Eq (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods (==) :: Heartbeat msg -> Heartbeat msg -> Bool Source # (/=) :: Heartbeat msg -> Heartbeat msg -> Bool Source #
Ord msg => Ord (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat Methods compare :: Heartbeat msg -> Heartbeat msg -> Ordering Source # (<) :: Heartbeat msg -> Heartbeat msg -> Bool Source # (<=) :: Heartbeat msg -> Heartbeat msg -> Bool Source # (>) :: Heartbeat msg -> Heartbeat msg -> Bool Source # (>=) :: Heartbeat msg -> Heartbeat msg -> Bool Source # max :: Heartbeat msg -> Heartbeat msg -> Heartbeat msg Source # min :: Heartbeat msg -> Heartbeat msg -> Heartbeat msg Source #
type Rep (Heartbeat msg) Source #
Instance details Defined in Hydra.Network.Heartbeat type Rep (Heartbeat msg) = D1 ('MetaData "Heartbeat" "Hydra.Network.Heartbeat" "hydra-node-0.20.0-GJlHI5giGamBfggh7PdYBV" 'False) (C1 ('MetaCons "Data" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 NodeId) :*: S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 msg)) :+: C1 ('MetaCons "Ping" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 NodeId)))

isPing :: Heartbeat msg -> Bool Source #

heartbeatDelay :: DiffTime Source #

Delay between each heartbeat check.

NOTE: This could be made configurable.

livenessDelay :: DiffTime Source #

Maximal delay between expected and sent heartbeats.

NOTE: This could be made configurable.

withHeartbeat Source #

Arguments

:: (MonadAsync m, MonadDelay m)
=> NodeId	This node's id, used to identify `Heartbeat` messages broadcast to peers.
-> NetworkComponent m (Heartbeat inbound) (Heartbeat outbound) a	Underlying `NetworkComponent` for sending and consuming `Heartbeat` messages.
-> NetworkComponent m (Either Connectivity inbound) outbound a	Returns a network component that can be used to send and consume arbitrary messages. This layer will take care of peeling out/wrapping messages into `Heartbeat`s.

Wrap a lower-level NetworkComponent and handle sending/receiving of heartbeats.

Note that the type of consumed and sent messages can be different.

updateStateFromIncomingMessages :: (MonadSTM m, MonadMonotonicTime m) => TVar m HeartbeatState -> NetworkCallback (Either Connectivity inbound) m -> NetworkCallback (Heartbeat inbound) m Source #

updateStateFromOutgoingMessages :: (MonadSTM m, MonadMonotonicTime m) => NodeId -> TVar m (Maybe Time) -> Network m (Heartbeat outbound) -> Network m outbound Source #

updateLastSent :: MonadSTM m => TVar m (Maybe Time) -> Time -> m () Source #

checkHeartbeatState :: (MonadDelay m, MonadSTM m) => NodeId -> TVar m (Maybe Time) -> Network m (Heartbeat outbound) -> m () Source #

shouldSendHeartbeat :: Time -> Maybe Time -> Bool Source #

checkRemoteParties :: (MonadDelay m, MonadSTM m) => TVar m HeartbeatState -> (Connectivity -> m ()) -> m () Source #

updateSuspected :: MonadSTM m => TVar m HeartbeatState -> Time -> m (Set NodeId) Source #

checkTimeout :: (DiffTime -> DiffTime) -> Time -> Time -> Bool Source #