{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}

-- | The node's LeiosNotify client and server, and what the node-to-node
-- wiring needs to run them.
module Ouroboros.Consensus.Network.NodeToNode.LeiosNotify
  ( leiosNotifyClient
  , leiosNotifyMiniProtocol
  , leiosNotifyPruner
  , leiosNotifyServer
  , withLeiosNotifyClientState
  ) where

import Cardano.Network.NodeToNode (addSafetyMargin)
import Cardano.Network.PeerSelection (PeerTrustable)
import Control.Monad.Except (Except, runExcept, runExceptT)
import Control.Tracer (Tracer, traceWith)
import Data.Foldable (for_)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Map.Strict (Map)
import qualified Data.Map.Strict as Map
import Data.Void (Void)
import Ouroboros.Consensus.Block
import Ouroboros.Consensus.BlockchainTime (BlockchainTime (..), SystemTime (..))
import Ouroboros.Consensus.Config (TopLevelConfig)
import Ouroboros.Consensus.HardFork.Abstract (HasHardForkHistory)
import Ouroboros.Consensus.Ledger.Abstract (GetTip, LedgerState, getTipSlot)
import Ouroboros.Consensus.Ledger.Extended (ledgerState)
import Ouroboros.Consensus.Ledger.SupportsProtocol (LedgerSupportsProtocol)
import Ouroboros.Consensus.Leios.AnnounceState (AnnounceState, pruneAnnounceState)
import Ouroboros.Consensus.Leios.NotifyLog
  ( Notification (..)
  , NotifyLog
  , nextNotifyLog
  , notifyLogStart
  )
import Ouroboros.Consensus.Leios.Types (LeiosPoint, LeiosVote (..))
import Ouroboros.Consensus.Leios.VoteState (AddVoteResult (..), VoteState, addVote)
import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck
  ( ClockSkew
  , SomeHeaderInFutureCheck
  )
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Client
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Client.Announcements.Validation
  ( AnnouncingHeader (ancHeader)
  )
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Client.State
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Server
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Trace (TraceLeiosNotifyClientEvent (..))
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Type
import Ouroboros.Consensus.Network.LeiosNotify
  ( leiosNotifyMiniProtocolNum
  , leiosNotifyPipelineDepth
  )
import Ouroboros.Consensus.Storage.ChainDB.API (ChainDB)
import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
import Ouroboros.Consensus.Util.IOLike
import Ouroboros.Network.ControlMessage (ControlMessageSTM)
import Ouroboros.Network.Mux
  ( MiniProtocol (..)
  , MiniProtocolLimits (..)
  , MiniProtocolWithExpandedCtx
  , RunMiniProtocolWithExpandedCtx
  , StartOnDemandOrEagerly (..)
  )

-- | The node's LeiosNotify client. It checks each reply against the peer's
-- state, which it updates, and disconnects the peer if a check fails.
leiosNotifyClient ::
  forall m blk.
  ( IOLike m
  , LedgerSupportsProtocol blk
  , ConvertRawHash blk
  , HasHardForkHistory blk
  ) =>
  Tracer m TraceLeiosNotifyClientEvent ->
  TopLevelConfig blk ->
  ChainDB m blk ->
  AnnounceState m blk ->
  VoteState m blk ->
  BlockchainTime m ->
  SystemTime m ->
  SomeHeaderInFutureCheck m blk ->
  ClockSkew ->
  ControlMessageSTM m ->
  StrictTVar m (LeiosNotifyPeerState blk) ->
  LeiosNotifyClientPeerPipelined LeiosPoint (Header blk) LeiosVote m ()
leiosNotifyClient :: forall (m :: * -> *) blk.
(IOLike m, LedgerSupportsProtocol blk, ConvertRawHash blk,
 HasHardForkHistory blk) =>
Tracer m TraceLeiosNotifyClientEvent
-> TopLevelConfig blk
-> ChainDB m blk
-> AnnounceState m blk
-> VoteState m blk
-> BlockchainTime m
-> SystemTime m
-> SomeHeaderInFutureCheck m blk
-> ClockSkew
-> ControlMessageSTM m
-> StrictTVar m (LeiosNotifyPeerState blk)
-> LeiosNotifyClientPeerPipelined
     LeiosPoint (Header blk) LeiosVote m ()
leiosNotifyClient
  Tracer m TraceLeiosNotifyClientEvent
tracer
  TopLevelConfig blk
cfg
  ChainDB m blk
chainDB
  AnnounceState m blk
announceState
  VoteState m blk
voteState
  BlockchainTime m
btime
  SystemTime m
systemTime
  SomeHeaderInFutureCheck m blk
chainSyncFutureCheck
  ClockSkew
maxClockSkew
  ControlMessageSTM m
controlMessageSTM
  StrictTVar m (LeiosNotifyPeerState blk)
stateVar =
    Int
-> STM m ()
-> ControlMessageSTM m
-> (Message
      (LeiosNotify LeiosPoint (Header blk) LeiosVote) 'StBusy 'StIdle
    -> m ())
-> LeiosNotifyClientPeerPipelined
     LeiosPoint (Header blk) LeiosVote m ()
forall (m :: * -> *) point announcement vote.
(PrimMonad m, MonadSTM m, MonadThrow m) =>
Int
-> STM m ()
-> ControlMessageSTM m
-> (Message (LeiosNotify point announcement vote) 'StBusy 'StIdle
    -> m ())
-> LeiosNotifyClientPeerPipelined point announcement vote m ()
leiosNotifyClientPeerHandlingReplies
      Int
leiosNotifyPipelineDepth
      ( TopLevelConfig blk
-> STM m (ExtLedgerState blk EmptyMK)
-> STM m CurrentSlot
-> STM m ()
forall (m :: * -> *) blk (mk :: MapKind).
(MonadSTM m, LedgerSupportsProtocol blk) =>
TopLevelConfig blk
-> STM m (ExtLedgerState blk mk) -> STM m CurrentSlot -> STM m ()
awaitImmTipCanForecastNow
          TopLevelConfig blk
cfg
          (ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
forall (m :: * -> *) blk.
ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
ChainDB.getImmutableLedger ChainDB m blk
chainDB)
          (BlockchainTime m -> STM m CurrentSlot
forall (m :: * -> *). BlockchainTime m -> STM m CurrentSlot
getCurrentSlot BlockchainTime m
btime)
      )
      ControlMessageSTM m
controlMessageSTM
      ( \case
          MsgLeiosBlockAnnouncement Header blk
hdr -> do
            (immLedger, st) <-
              STM m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk)
-> m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk)
 -> m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk))
-> STM m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk)
-> m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk)
forall a b. (a -> b) -> a -> b
$ (,) (ExtLedgerState blk EmptyMK
 -> LeiosNotifyPeerState blk
 -> (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk))
-> STM m (ExtLedgerState blk EmptyMK)
-> STM
     m
     (LeiosNotifyPeerState blk
      -> (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
forall (m :: * -> *) blk.
ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
ChainDB.getImmutableLedger ChainDB m blk
chainDB STM
  m
  (LeiosNotifyPeerState blk
   -> (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk))
-> STM m (LeiosNotifyPeerState blk)
-> STM m (ExtLedgerState blk EmptyMK, LeiosNotifyPeerState blk)
forall a b. STM m (a -> b) -> STM m a -> STM m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> StrictTVar m (LeiosNotifyPeerState blk)
-> STM m (LeiosNotifyPeerState blk)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (LeiosNotifyPeerState blk)
stateVar
            st' <-
              either throwIO pure
                =<< runExceptT
                  ( onLeiosNotifyAnnouncement
                      systemTime
                      chainSyncFutureCheck
                      cfg
                      announceState
                      immLedger
                      st
                      hdr
                  )
            -- The pruner may have run meanwhile; keep its work, so that every entry
            -- is still at or after 'lnpsPruneSlot'.
            atomically $ do
              pruneSlot <- lnpsPruneSlot <$> readTVar stateVar
              writeTVar stateVar $! pruneLeiosNotifyPeerState pruneSlot st'
          MsgLeiosBlockOffer LeiosPoint
point Word32
claimed ->
            STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
              immLedger <- ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
forall (m :: * -> *) blk.
ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
ChainDB.getImmutableLedger ChainDB m blk
chainDB
              checkAndUpdate_ $ checkLeiosBlockOffer cfg immLedger point claimed
          MsgLeiosBlockTxsOffer LeiosPoint
point ->
            STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ (LeiosNotifyPeerState blk
 -> Except ExnLeiosNotifyInvalidOffer (LeiosNotifyPeerState blk))
-> STM m ()
forall e.
Exception e =>
(LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk))
-> STM m ()
checkAndUpdate_ ((LeiosNotifyPeerState blk
  -> Except ExnLeiosNotifyInvalidOffer (LeiosNotifyPeerState blk))
 -> STM m ())
-> (LeiosNotifyPeerState blk
    -> Except ExnLeiosNotifyInvalidOffer (LeiosNotifyPeerState blk))
-> STM m ()
forall a b. (a -> b) -> a -> b
$ LeiosPoint
-> LeiosNotifyPeerState blk
-> Except ExnLeiosNotifyInvalidOffer (LeiosNotifyPeerState blk)
forall blk.
LeiosPoint
-> LeiosNotifyPeerState blk
-> Except ExnLeiosNotifyInvalidOffer (LeiosNotifyPeerState blk)
checkLeiosClosureOffer LeiosPoint
point
          MsgLeiosVotes NonEmpty LeiosVote
votes -> do
            now <- SystemTime m -> m RelativeTime
forall (m :: * -> *). SystemTime m -> m RelativeTime
systemTimeCurrent SystemTime m
systemTime
            accepted <- atomically $ do
              immLedger <- ChainDB.getImmutableLedger chainDB
              checkAndUpdate $ checkLeiosVotes cfg maxClockSkew immLedger now votes
            for_ accepted $ \vote :: AcceptedVote
vote@MkAcceptedVote{LeiosVote
acceptedVote :: LeiosVote
acceptedVote :: AcceptedVote -> LeiosVote
acceptedVote} ->
              VoteState m blk -> SystemTime m -> AcceptedVote -> m AddVoteResult
forall (m :: * -> *) blk.
(IOLike m, ConvertRawHash blk, GetHeader blk) =>
VoteState m blk -> SystemTime m -> AcceptedVote -> m AddVoteResult
addVote VoteState m blk
voteState SystemTime m
systemTime AcceptedVote
vote m AddVoteResult -> (AddVoteResult -> 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
>>= \case
                AddVoteResult
Added ->
                  Tracer m TraceLeiosNotifyClientEvent
-> TraceLeiosNotifyClientEvent -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m TraceLeiosNotifyClientEvent
tracer (TraceLeiosNotifyClientEvent -> m ())
-> TraceLeiosNotifyClientEvent -> m ()
forall a b. (a -> b) -> a -> b
$ LeiosVote -> TraceLeiosNotifyClientEvent
TraceLeiosNotifyClientVoteAdded LeiosVote
acceptedVote
                VoteInvalid InvalidVoteReason
reason ->
                  ExnLeiosNotifyInvalidVotes -> m ()
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (ExnLeiosNotifyInvalidVotes -> m ())
-> ExnLeiosNotifyInvalidVotes -> m ()
forall a b. (a -> b) -> a -> b
$ AcceptedVote -> InvalidVoteReason -> ExnLeiosNotifyInvalidVotes
ExnLeiosInvalidVote AcceptedVote
vote InvalidVoteReason
reason
                AddVoteResult
ArrivedTooLate ->
                  Tracer m TraceLeiosNotifyClientEvent
-> TraceLeiosNotifyClientEvent -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m TraceLeiosNotifyClientEvent
tracer (TraceLeiosNotifyClientEvent -> m ())
-> TraceLeiosNotifyClientEvent -> m ()
forall a b. (a -> b) -> a -> b
$ LeiosVote -> TraceLeiosNotifyClientEvent
TraceLeiosNotifyClientVoteArrivedTooLate LeiosVote
acceptedVote
                AddVoteResult
AlreadyKnown -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                AddVoteResult
NotAnnounced -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          Message
  (LeiosNotify LeiosPoint (Header blk) LeiosVote) 'StBusy 'StIdle
R:MessageLeiosNotifyfromto
  LeiosPoint (Header blk) LeiosVote 'StBusy 'StIdle
MsgCanceled -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      )
   where
    checkAndUpdate ::
      Exception e =>
      (LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk, a)) ->
      STM m a
    checkAndUpdate :: forall e a.
Exception e =>
(LeiosNotifyPeerState blk
 -> Except e (LeiosNotifyPeerState blk, a))
-> STM m a
checkAndUpdate LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk, a)
judge = do
      result <- StrictTVar m (LeiosNotifyPeerState blk)
-> (LeiosNotifyPeerState blk
    -> (Either e a, LeiosNotifyPeerState blk))
-> STM m (Either e a)
forall (m :: * -> *) s a.
MonadSTM m =>
StrictTVar m s -> (s -> (a, s)) -> STM m a
stateTVar StrictTVar m (LeiosNotifyPeerState blk)
stateVar ((LeiosNotifyPeerState blk
  -> (Either e a, LeiosNotifyPeerState blk))
 -> STM m (Either e a))
-> (LeiosNotifyPeerState blk
    -> (Either e a, LeiosNotifyPeerState blk))
-> STM m (Either e a)
forall a b. (a -> b) -> a -> b
$ \LeiosNotifyPeerState blk
st -> case Except e (LeiosNotifyPeerState blk, a)
-> Either e (LeiosNotifyPeerState blk, a)
forall e a. Except e a -> Either e a
runExcept (LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk, a)
judge LeiosNotifyPeerState blk
st) of
        Left e
e -> (e -> Either e a
forall a b. a -> Either a b
Left e
e, LeiosNotifyPeerState blk
st)
        Right (LeiosNotifyPeerState blk
st', a
a) -> (a -> Either e a
forall a b. b -> Either a b
Right a
a, LeiosNotifyPeerState blk
st')
      either throwSTM pure result

    checkAndUpdate_ ::
      Exception e =>
      (LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk)) ->
      STM m ()
    checkAndUpdate_ :: forall e.
Exception e =>
(LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk))
-> STM m ()
checkAndUpdate_ LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk)
judge = (LeiosNotifyPeerState blk
 -> Except e (LeiosNotifyPeerState blk, ()))
-> STM m ()
forall e a.
Exception e =>
(LeiosNotifyPeerState blk
 -> Except e (LeiosNotifyPeerState blk, a))
-> STM m a
checkAndUpdate ((LeiosNotifyPeerState blk -> (LeiosNotifyPeerState blk, ()))
-> Except e (LeiosNotifyPeerState blk)
-> Except e (LeiosNotifyPeerState blk, ())
forall a b.
(a -> b) -> ExceptT e Identity a -> ExceptT e Identity b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (,()) (Except e (LeiosNotifyPeerState blk)
 -> Except e (LeiosNotifyPeerState blk, ()))
-> (LeiosNotifyPeerState blk
    -> Except e (LeiosNotifyPeerState blk))
-> LeiosNotifyPeerState blk
-> Except e (LeiosNotifyPeerState blk, ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk)
judge)

-- | The node's LeiosNotify server, which relays the node's 'NotifyLog' to one
-- downstream peer, from the log's oldest notification on.
leiosNotifyServer ::
  IOLike m =>
  NotifyLog m blk ->
  m (LeiosNotifyServerPeerLookahead LeiosPoint (Header blk) LeiosVote m)
leiosNotifyServer :: forall (m :: * -> *) blk.
IOLike m =>
NotifyLog m blk
-> m (LeiosNotifyServerPeerLookahead
        LeiosPoint (Header blk) LeiosVote m)
leiosNotifyServer NotifyLog m blk
notifyLog = do
  requests <- Int -> m (StrictTVar m Int)
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> m (StrictTVar m a)
newTVarIO (Int
0 :: Int)
  cursorVar <- newTVarIO notifyLogStart
  leiosNotifyServerPeerLookahead
    (increment requests)
    (serveNext requests cursorVar)
 where
  increment :: StrictTVar m Int -> m WhetherExcessiveRequests
increment StrictTVar m Int
requests = STM m WhetherExcessiveRequests -> m WhetherExcessiveRequests
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m WhetherExcessiveRequests -> m WhetherExcessiveRequests)
-> STM m WhetherExcessiveRequests -> m WhetherExcessiveRequests
forall a b. (a -> b) -> a -> b
$ do
    n <- StrictTVar m Int -> STM m Int
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m Int
requests
    if n >= leiosNotifyPipelineDepth
      then pure ExcessiveRequests
      else do
        writeTVar requests (n + 1)
        pure NotExcessiveRequests

  serveNext :: StrictTVar m Int
-> StrictTVar m NotifyCursor
-> STM
     m
     (Message
        (LeiosNotify LeiosPoint (Header blk) LeiosVote) 'StBusy 'StIdle)
serveNext StrictTVar m Int
requests StrictTVar m NotifyCursor
cursorVar = do
    -- TODO: flush the log ahead cursor instead
    (n, cursor) <- NotifyLog m blk
-> NotifyCursor -> STM m (Notification blk, NotifyCursor)
forall (m :: * -> *) blk.
IOLike m =>
NotifyLog m blk
-> NotifyCursor -> STM m (Notification blk, NotifyCursor)
nextNotifyLog NotifyLog m blk
notifyLog (NotifyCursor -> STM m (Notification blk, NotifyCursor))
-> STM m NotifyCursor -> STM m (Notification blk, NotifyCursor)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< StrictTVar m NotifyCursor -> STM m NotifyCursor
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m NotifyCursor
cursorVar
    writeTVar cursorVar cursor
    modifyTVar requests (subtract 1)
    pure $ case n of
      NotifyAnnouncement AnnouncingHeader blk
anc -> Header blk
-> Message
     (LeiosNotify LeiosPoint (Header blk) LeiosVote) 'StBusy 'StIdle
forall announcement point vote.
announcement
-> Message (LeiosNotify point announcement vote) 'StBusy 'StIdle
MsgLeiosBlockAnnouncement (AnnouncingHeader blk -> Header blk
forall blk. AnnouncingHeader blk -> Header blk
ancHeader AnnouncingHeader blk
anc)
      -- One vote per message: the client pipelines its requests, so several
      -- messages cost about what several votes in one would.
      NotifyVote LeiosVote
v -> NonEmpty LeiosVote
-> Message
     (LeiosNotify LeiosPoint (Header blk) LeiosVote) 'StBusy 'StIdle
forall vote point announcement.
NonEmpty vote
-> Message (LeiosNotify point announcement vote) 'StBusy 'StIdle
MsgLeiosVotes (LeiosVote
v LeiosVote -> [LeiosVote] -> NonEmpty LeiosVote
forall a. a -> [a] -> NonEmpty a
:| [])

-- | Run the given action with a fresh LeiosNotify client state, registered
-- under the peer for as long as the action runs.
withLeiosNotifyClientState ::
  (IOLike m, Ord peer, NoThunks (Header blk)) =>
  StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk))) ->
  peer ->
  (StrictTVar m (LeiosNotifyPeerState blk) -> m a) ->
  m a
withLeiosNotifyClientState :: forall (m :: * -> *) peer blk a.
(IOLike m, Ord peer, NoThunks (Header blk)) =>
StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
-> peer -> (StrictTVar m (LeiosNotifyPeerState blk) -> m a) -> m a
withLeiosNotifyClientState StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
statesVar peer
peer =
  m (StrictTVar m (LeiosNotifyPeerState blk))
-> (StrictTVar m (LeiosNotifyPeerState blk) -> m ())
-> (StrictTVar m (LeiosNotifyPeerState blk) -> m a)
-> m a
forall a b c. m a -> (a -> m b) -> (a -> m c) -> m c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket
    ( STM m (StrictTVar m (LeiosNotifyPeerState blk))
-> m (StrictTVar m (LeiosNotifyPeerState blk))
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (StrictTVar m (LeiosNotifyPeerState blk))
 -> m (StrictTVar m (LeiosNotifyPeerState blk)))
-> STM m (StrictTVar m (LeiosNotifyPeerState blk))
-> m (StrictTVar m (LeiosNotifyPeerState blk))
forall a b. (a -> b) -> a -> b
$ do
        stateVar <- LeiosNotifyPeerState blk
-> STM m (StrictTVar m (LeiosNotifyPeerState blk))
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> STM m (StrictTVar m a)
newTVar LeiosNotifyPeerState blk
forall blk. LeiosNotifyPeerState blk
emptyLeiosNotifyPeerState
        modifyTVar statesVar (Map.insert peer stateVar)
        pure stateVar
    )
    (\StrictTVar m (LeiosNotifyPeerState blk)
_stateVar -> STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
-> (Map peer (StrictTVar m (LeiosNotifyPeerState blk))
    -> Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
-> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
statesVar (peer
-> Map peer (StrictTVar m (LeiosNotifyPeerState blk))
-> Map peer (StrictTVar m (LeiosNotifyPeerState blk))
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete peer
peer))

-- | Prune every LeiosNotify client's state, and the node's 'AnnounceState',
-- each time the immutable tip's slot advances.
leiosNotifyPruner ::
  ( IOLike m
  , GetTip (LedgerState blk)
  , ConvertRawHash blk
  , GetHeader blk
  ) =>
  ChainDB m blk ->
  AnnounceState m blk ->
  StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk))) ->
  m Void
leiosNotifyPruner :: forall (m :: * -> *) blk peer.
(IOLike m, GetTip (LedgerState blk), ConvertRawHash blk,
 GetHeader blk) =>
ChainDB m blk
-> AnnounceState m blk
-> StrictTVar
     m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
-> m Void
leiosNotifyPruner ChainDB m blk
chainDB AnnounceState m blk
announceState StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
statesVar = SlotNo -> m Void
go (Word64 -> SlotNo
SlotNo Word64
0)
 where
  go :: SlotNo -> m Void
go SlotNo
prunedSlot = do
    immTipSlot <- STM m SlotNo -> m SlotNo
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m SlotNo -> m SlotNo) -> STM m SlotNo -> m SlotNo
forall a b. (a -> b) -> a -> b
$ do
      immLedger <- ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
forall (m :: * -> *) blk.
ChainDB m blk -> STM m (ExtLedgerState blk EmptyMK)
ChainDB.getImmutableLedger ChainDB m blk
chainDB
      case getTipSlot (ledgerState immLedger) of
        NotOrigin SlotNo
slot
          | SlotNo
prunedSlot SlotNo -> SlotNo -> Bool
forall a. Ord a => a -> a -> Bool
< SlotNo
slot -> do
              states <- StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
-> STM m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
statesVar
              mapM_ (\StrictTVar m (LeiosNotifyPeerState blk)
stateVar -> StrictTVar m (LeiosNotifyPeerState blk)
-> (LeiosNotifyPeerState blk -> LeiosNotifyPeerState blk)
-> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m (LeiosNotifyPeerState blk)
stateVar (SlotNo -> LeiosNotifyPeerState blk -> LeiosNotifyPeerState blk
forall blk.
(ConvertRawHash blk, GetHeader blk) =>
SlotNo -> LeiosNotifyPeerState blk -> LeiosNotifyPeerState blk
pruneLeiosNotifyPeerState SlotNo
slot)) states
              pruneAnnounceState announceState slot
              pure slot
        WithOrigin SlotNo
_ -> STM m SlotNo
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry
    go immTipSlot

-- | LeiosNotify as a node-to-node mini-protocol.
leiosNotifyMiniProtocol ::
  RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c ->
  MiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
leiosNotifyMiniProtocol :: forall (mode :: Mode) addr b (m :: * -> *) a c.
RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
-> MiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
leiosNotifyMiniProtocol RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
run =
  MiniProtocol
    { miniProtocolNum :: MiniProtocolNum
miniProtocolNum = MiniProtocolNum
leiosNotifyMiniProtocolNum
    , miniProtocolStart :: StartOnDemandOrEagerly
miniProtocolStart = StartOnDemandOrEagerly
StartOnDemand
    , miniProtocolLimits :: MiniProtocolLimits
miniProtocolLimits = MiniProtocolLimits
leiosNotifyProtocolLimits
    , miniProtocolRun :: RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
miniProtocolRun = RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
run
    }

-- | The ingress queue must hold a reply to every request the client may leave
-- outstanding, each as large as 'maxLeiosNotifyReplyBytes' allows; the
-- server's, which only receives requests, needs far less.
leiosNotifyProtocolLimits :: MiniProtocolLimits
leiosNotifyProtocolLimits :: MiniProtocolLimits
leiosNotifyProtocolLimits =
  MiniProtocolLimits
    { maximumIngressQueue :: Int
maximumIngressQueue =
        Int -> Int
addSafetyMargin (Int -> Int) -> Int -> Int
forall a b. (a -> b) -> a -> b
$ Int
leiosNotifyPipelineDepth Int -> Int -> Int
forall a. Num a => a -> a -> a
* Word -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word
maxLeiosNotifyReplyBytes
    }