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

-- | 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
  , leiosNotifyProtocols
  , leiosNotifyPruner
  , leiosNotifyServer
  , withLeiosNotifyClientState
  ) where

import Cardano.Network.NodeToNode (NodeToNodeVersion (..), addSafetyMargin)
import Cardano.Network.PeerSelection (PeerTrustable)
import Control.Monad.Except (Except, runExcept, runExceptT)
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.Protocol (LeiosProtocol)
import Ouroboros.Consensus.Leios.Types (LeiosPoint, LeiosVote)
import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck
  ( ClockSkew
  , SomeHeaderInFutureCheck
  )
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Client
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Client.State
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Server
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 (..)
  , OuroborosBundleWithExpandedCtx
  , RunMiniProtocolWithExpandedCtx
  , StartOnDemandOrEagerly (..)
  , WithProtocolTemperature (..)
  , withHot
  )

-- | 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
  , LeiosProtocol (BlockProtocol blk)
  , ConvertRawHash blk
  , HasHardForkHistory blk
  ) =>
  TopLevelConfig blk ->
  ChainDB 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,
 LeiosProtocol (BlockProtocol blk), ConvertRawHash blk,
 HasHardForkHistory blk) =>
TopLevelConfig blk
-> ChainDB m blk
-> BlockchainTime m
-> SystemTime m
-> SomeHeaderInFutureCheck m blk
-> ClockSkew
-> ControlMessageSTM m
-> StrictTVar m (LeiosNotifyPeerState blk)
-> LeiosNotifyClientPeerPipelined
     LeiosPoint (Header blk) LeiosVote m ()
leiosNotifyClient
  TopLevelConfig blk
cfg
  ChainDB m blk
chainDB
  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
                      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
            atomically $ do
              immLedger <- ChainDB.getImmutableLedger chainDB
              checkAndUpdate $ checkLeiosVotes cfg maxClockSkew immLedger now votes
          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)) ->
      STM m ()
    checkAndUpdate :: forall e.
Exception e =>
(LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk))
-> STM m ()
checkAndUpdate LeiosNotifyPeerState blk -> Except e (LeiosNotifyPeerState blk)
judge = do
      st <- 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
      either throwSTM (writeTVar stateVar) $ runExcept (judge st)

-- | The node's LeiosNotify server.
leiosNotifyServer ::
  IOLike m =>
  m (LeiosNotifyServerPeerLookahead LeiosPoint (Header blk) LeiosVote m)
leiosNotifyServer :: forall (m :: * -> *) blk.
IOLike m =>
m (LeiosNotifyServerPeerLookahead
     LeiosPoint (Header blk) LeiosVote m)
leiosNotifyServer = do
  requests <- Int -> m (StrictTVar m Int)
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> m (StrictTVar m a)
newTVarIO (Int
0 :: Int)
  leiosNotifyServerPeerLookahead
    ( atomically $ do
        n <- readTVar requests
        if n >= leiosNotifyPipelineDepth
          then pure ExcessiveRequests
          else do
            writeTVar requests (n + 1)
            pure NotExcessiveRequests
    )
    -- The node never has a notification to send.
    retry

-- | 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 each time the immutable tip's slot
-- advances.
leiosNotifyPruner ::
  ( IOLike m
  , GetTip (LedgerState blk)
  , ConvertRawHash blk
  , GetHeader blk
  ) =>
  ChainDB 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
-> StrictTVar
     m (Map peer (StrictTVar m (LeiosNotifyPeerState blk)))
-> m Void
leiosNotifyPruner ChainDB m blk
chainDB 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
              pure slot
        WithOrigin SlotNo
_ -> STM m SlotNo
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry
    go immTipSlot

-- | LeiosNotify, to add to a node-to-node bundle, if the version has it.
--
-- TODO remove once 'NodeToNodeProtocols' has fields for the Leios mini-protocols.
leiosNotifyProtocols ::
  NodeToNodeVersion ->
  RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c ->
  OuroborosBundleWithExpandedCtx mode addr PeerTrustable b m a c
leiosNotifyProtocols :: forall (mode :: Mode) addr b (m :: * -> *) a c.
NodeToNodeVersion
-> RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
-> OuroborosBundleWithExpandedCtx mode addr PeerTrustable b m a c
leiosNotifyProtocols NodeToNodeVersion
version RunMiniProtocolWithExpandedCtx mode addr PeerTrustable b m a c
run
  | NodeToNodeVersion
version NodeToNodeVersion -> NodeToNodeVersion -> Bool
forall a. Ord a => a -> a -> Bool
>= NodeToNodeVersion
NodeToNodeV_16 =
      TemperatureBundle
  [MiniProtocol
     mode
     (ExpandedInitiatorContext addr PeerTrustable m)
     (ResponderContext addr)
     b
     m
     a
     c]
forall a. Monoid a => a
mempty
        { withHot =
            WithHot
              [ MiniProtocol
                  { miniProtocolNum = leiosNotifyMiniProtocolNum
                  , miniProtocolStart = StartOnDemand
                  , miniProtocolLimits = leiosNotifyProtocolLimits
                  , miniProtocolRun = run
                  }
              ]
        }
  | Bool
otherwise = TemperatureBundle
  [MiniProtocol
     mode
     (ExpandedInitiatorContext addr PeerTrustable m)
     (ResponderContext addr)
     b
     m
     a
     c]
forall a. Monoid a => a
mempty

-- | 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
    }