{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# LANGUAGE UndecidableInstances #-}

-- | What the node relays to its downstream peers over LeiosNotify.
--
-- The node keeps one log of notifications, in the order it appended them, and
-- every LeiosNotify server reads that log through its own 'NotifyCursor'. So a
-- notification is stored once however many peers it goes to, and a peer that
-- requests slowly merely lags behind in the log, rather than piling up a queue
-- of its own.
--
-- Whoever appends a notification says until when it is worth relaying; the
-- pruner drops it then, and the log does not take one it was pruned past
-- already. So the node relays none longer than 'notifyLogPruneCadence' after
-- that. A peer that connects
-- late starts at the log's oldest notification, so it still receives
-- everything worth relaying, and a notification is never worth relaying longer
-- than the announcement it is about, so a peer always receives the
-- announcement first. A cursor that falls behind pruning merely skips what was
-- pruned.
module Ouroboros.Consensus.Leios.NotifyLog
  ( Notification (..)
  , NotifyCursor
  , NotifyLog
  , appendNotifyLog
  , newNotifyLog
  , nextNotifyLog
  , notifyLogPruneCadence
  , notifyLogPruner
  , notifyLogStart
  , pruneNotifyLog
  ) where

import Control.Monad (forever, unless)
import Data.Map.Strict (Map)
import qualified Data.Map.Strict as Map
import qualified Data.Set as Set
import Data.Time.Clock (NominalDiffTime)
import Data.Void (Void)
import Data.Word (Word64)
import GHC.Generics (Generic)
import NoThunks.Class (NoThunks)
import Ouroboros.Consensus.Block (Header)
import Ouroboros.Consensus.BlockchainTime.WallClock.Types
  ( RelativeTime (..)
  , SystemTime (..)
  )
import Ouroboros.Consensus.Leios.Types (LeiosVote)
import Ouroboros.Consensus.MiniProtocol.LeiosNotify.Client.Announcements.Validation
  ( AnnouncingHeader
  )
import Ouroboros.Consensus.Util.IOLike
  ( IOLike
  , MonadDelay (threadDelay)
  , STM
  , StrictTVar
  , atomically
  , modifyTVar
  , newTVarIO
  , readTVar
  , retry
  , writeTVar
  )
import Ouroboros.Consensus.Util.Time (nominalDelay)

-- | A notification the node relays.
data Notification blk
  = NotifyAnnouncement !(AnnouncingHeader blk)
  | NotifyVote !LeiosVote
  -- TODO: add offers
  deriving stock (forall x. Notification blk -> Rep (Notification blk) x)
-> (forall x. Rep (Notification blk) x -> Notification blk)
-> Generic (Notification blk)
forall x. Rep (Notification blk) x -> Notification blk
forall x. Notification blk -> Rep (Notification blk) x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
forall blk x. Rep (Notification blk) x -> Notification blk
forall blk x. Notification blk -> Rep (Notification blk) x
$cfrom :: forall blk x. Notification blk -> Rep (Notification blk) x
from :: forall x. Notification blk -> Rep (Notification blk) x
$cto :: forall blk x. Rep (Notification blk) x -> Notification blk
to :: forall x. Rep (Notification blk) x -> Notification blk
Generic

deriving anyclass instance NoThunks (Header blk) => NoThunks (Notification blk)

-- | The node's log of notifications to relay.
data NotifyLog m blk = MkNotifyLog
  { forall (m :: * -> *) blk.
NotifyLog m blk
-> StrictTVar m (Map NotifyCursor (Notification blk))
notifyEntries :: !(StrictTVar m (Map NotifyCursor (Notification blk)))
  -- ^ Each notification, at the position it was appended at.
  , forall (m :: * -> *) blk.
NotifyLog m blk -> StrictTVar m NotifyCursor
notifyEnd :: !(StrictTVar m NotifyCursor)
  -- ^ Where the next notification will be appended.
  , forall (m :: * -> *) blk.
NotifyLog m blk -> StrictTVar m (Map RelativeTime [NotifyCursor])
notifyExpiry :: !(StrictTVar m (Map RelativeTime [NotifyCursor]))
  -- ^ The position of each notification, by until when it is worth relaying,
  -- so that pruning visits only what expires.
  , forall (m :: * -> *) blk.
NotifyLog m blk -> StrictTVar m RelativeTime
notifyPrunedAt :: !(StrictTVar m RelativeTime)
  -- ^ When the log was last pruned.
  }

-- | A position in the log, which only ever grows.
newtype NotifyCursor = MkNotifyCursor Word64
  deriving stock (NotifyCursor -> NotifyCursor -> Bool
(NotifyCursor -> NotifyCursor -> Bool)
-> (NotifyCursor -> NotifyCursor -> Bool) -> Eq NotifyCursor
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: NotifyCursor -> NotifyCursor -> Bool
== :: NotifyCursor -> NotifyCursor -> Bool
$c/= :: NotifyCursor -> NotifyCursor -> Bool
/= :: NotifyCursor -> NotifyCursor -> Bool
Eq, Eq NotifyCursor
Eq NotifyCursor =>
(NotifyCursor -> NotifyCursor -> Ordering)
-> (NotifyCursor -> NotifyCursor -> Bool)
-> (NotifyCursor -> NotifyCursor -> Bool)
-> (NotifyCursor -> NotifyCursor -> Bool)
-> (NotifyCursor -> NotifyCursor -> Bool)
-> (NotifyCursor -> NotifyCursor -> NotifyCursor)
-> (NotifyCursor -> NotifyCursor -> NotifyCursor)
-> Ord NotifyCursor
NotifyCursor -> NotifyCursor -> Bool
NotifyCursor -> NotifyCursor -> Ordering
NotifyCursor -> NotifyCursor -> NotifyCursor
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: NotifyCursor -> NotifyCursor -> Ordering
compare :: NotifyCursor -> NotifyCursor -> Ordering
$c< :: NotifyCursor -> NotifyCursor -> Bool
< :: NotifyCursor -> NotifyCursor -> Bool
$c<= :: NotifyCursor -> NotifyCursor -> Bool
<= :: NotifyCursor -> NotifyCursor -> Bool
$c> :: NotifyCursor -> NotifyCursor -> Bool
> :: NotifyCursor -> NotifyCursor -> Bool
$c>= :: NotifyCursor -> NotifyCursor -> Bool
>= :: NotifyCursor -> NotifyCursor -> Bool
$cmax :: NotifyCursor -> NotifyCursor -> NotifyCursor
max :: NotifyCursor -> NotifyCursor -> NotifyCursor
$cmin :: NotifyCursor -> NotifyCursor -> NotifyCursor
min :: NotifyCursor -> NotifyCursor -> NotifyCursor
Ord, Int -> NotifyCursor -> ShowS
[NotifyCursor] -> ShowS
NotifyCursor -> String
(Int -> NotifyCursor -> ShowS)
-> (NotifyCursor -> String)
-> ([NotifyCursor] -> ShowS)
-> Show NotifyCursor
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> NotifyCursor -> ShowS
showsPrec :: Int -> NotifyCursor -> ShowS
$cshow :: NotifyCursor -> String
show :: NotifyCursor -> String
$cshowList :: [NotifyCursor] -> ShowS
showList :: [NotifyCursor] -> ShowS
Show)
  deriving newtype Context -> NotifyCursor -> IO (Maybe ThunkInfo)
Proxy NotifyCursor -> String
(Context -> NotifyCursor -> IO (Maybe ThunkInfo))
-> (Context -> NotifyCursor -> IO (Maybe ThunkInfo))
-> (Proxy NotifyCursor -> String)
-> NoThunks NotifyCursor
forall a.
(Context -> a -> IO (Maybe ThunkInfo))
-> (Context -> a -> IO (Maybe ThunkInfo))
-> (Proxy a -> String)
-> NoThunks a
$cnoThunks :: Context -> NotifyCursor -> IO (Maybe ThunkInfo)
noThunks :: Context -> NotifyCursor -> IO (Maybe ThunkInfo)
$cwNoThunks :: Context -> NotifyCursor -> IO (Maybe ThunkInfo)
wNoThunks :: Context -> NotifyCursor -> IO (Maybe ThunkInfo)
$cshowTypeOf :: Proxy NotifyCursor -> String
showTypeOf :: Proxy NotifyCursor -> String
NoThunks

-- | Where a downstream peer starts reading: before every notification,
-- including the ones already pruned.
notifyLogStart :: NotifyCursor
notifyLogStart :: NotifyCursor
notifyLogStart = Word64 -> NotifyCursor
MkNotifyCursor Word64
0

newNotifyLog :: (IOLike m, NoThunks (Header blk)) => m (NotifyLog m blk)
newNotifyLog :: forall (m :: * -> *) blk.
(IOLike m, NoThunks (Header blk)) =>
m (NotifyLog m blk)
newNotifyLog =
  StrictTVar m (Map NotifyCursor (Notification blk))
-> StrictTVar m NotifyCursor
-> StrictTVar m (Map RelativeTime [NotifyCursor])
-> StrictTVar m RelativeTime
-> NotifyLog m blk
forall (m :: * -> *) blk.
StrictTVar m (Map NotifyCursor (Notification blk))
-> StrictTVar m NotifyCursor
-> StrictTVar m (Map RelativeTime [NotifyCursor])
-> StrictTVar m RelativeTime
-> NotifyLog m blk
MkNotifyLog
    (StrictTVar m (Map NotifyCursor (Notification blk))
 -> StrictTVar m NotifyCursor
 -> StrictTVar m (Map RelativeTime [NotifyCursor])
 -> StrictTVar m RelativeTime
 -> NotifyLog m blk)
-> m (StrictTVar m (Map NotifyCursor (Notification blk)))
-> m (StrictTVar m NotifyCursor
      -> StrictTVar m (Map RelativeTime [NotifyCursor])
      -> StrictTVar m RelativeTime
      -> NotifyLog m blk)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Map NotifyCursor (Notification blk)
-> m (StrictTVar m (Map NotifyCursor (Notification blk)))
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> m (StrictTVar m a)
newTVarIO Map NotifyCursor (Notification blk)
forall k a. Map k a
Map.empty
    m (StrictTVar m NotifyCursor
   -> StrictTVar m (Map RelativeTime [NotifyCursor])
   -> StrictTVar m RelativeTime
   -> NotifyLog m blk)
-> m (StrictTVar m NotifyCursor)
-> m (StrictTVar m (Map RelativeTime [NotifyCursor])
      -> StrictTVar m RelativeTime -> NotifyLog m blk)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> NotifyCursor -> m (StrictTVar m NotifyCursor)
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> m (StrictTVar m a)
newTVarIO NotifyCursor
notifyLogStart
    m (StrictTVar m (Map RelativeTime [NotifyCursor])
   -> StrictTVar m RelativeTime -> NotifyLog m blk)
-> m (StrictTVar m (Map RelativeTime [NotifyCursor]))
-> m (StrictTVar m RelativeTime -> NotifyLog m blk)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Map RelativeTime [NotifyCursor]
-> m (StrictTVar m (Map RelativeTime [NotifyCursor]))
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> m (StrictTVar m a)
newTVarIO Map RelativeTime [NotifyCursor]
forall k a. Map k a
Map.empty
    m (StrictTVar m RelativeTime -> NotifyLog m blk)
-> m (StrictTVar m RelativeTime) -> m (NotifyLog m blk)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> RelativeTime -> m (StrictTVar m RelativeTime)
forall (m :: * -> *) a.
(HasCallStack, MonadSTM m, NoThunks a) =>
a -> m (StrictTVar m a)
newTVarIO (NominalDiffTime -> RelativeTime
RelativeTime NominalDiffTime
0)

-- | Append a notification worth relaying until the given time, unless the log
-- was pruned past it already.
--
-- After that time, the notification is of no use to a downstream peer: a vote's
-- deadline, or when an announcement is too old to relay. Either leaves a
-- grace, for transmission and queueing, before the peer would reject it.
appendNotifyLog ::
  IOLike m =>
  NotifyLog m blk -> Notification blk -> RelativeTime -> STM m ()
appendNotifyLog :: forall (m :: * -> *) blk.
IOLike m =>
NotifyLog m blk -> Notification blk -> RelativeTime -> STM m ()
appendNotifyLog NotifyLog m blk
notifyLog Notification blk
n RelativeTime
expires = do
  prunedAt <- StrictTVar m RelativeTime -> STM m RelativeTime
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar (NotifyLog m blk -> StrictTVar m RelativeTime
forall (m :: * -> *) blk.
NotifyLog m blk -> StrictTVar m RelativeTime
notifyPrunedAt NotifyLog m blk
notifyLog)
  unless (expires < prunedAt) $ do
    end <- readTVar (notifyEnd notifyLog)
    modifyTVar (notifyEntries notifyLog) (Map.insert end n)
    modifyTVar (notifyExpiry notifyLog) (Map.insertWith (<>) expires [end])
    writeTVar (notifyEnd notifyLog) (succCursor end)

-- | The first notification at or after the cursor, and the cursor just past
-- it.
--
-- Retries until there is such a notification.
nextNotifyLog ::
  IOLike m =>
  NotifyLog m blk ->
  NotifyCursor ->
  STM m (Notification blk, NotifyCursor)
nextNotifyLog :: forall (m :: * -> *) blk.
IOLike m =>
NotifyLog m blk
-> NotifyCursor -> STM m (Notification blk, NotifyCursor)
nextNotifyLog NotifyLog m blk
notifyLog NotifyCursor
cursor = do
  entries <- StrictTVar m (Map NotifyCursor (Notification blk))
-> STM m (Map NotifyCursor (Notification blk))
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar (NotifyLog m blk
-> StrictTVar m (Map NotifyCursor (Notification blk))
forall (m :: * -> *) blk.
NotifyLog m blk
-> StrictTVar m (Map NotifyCursor (Notification blk))
notifyEntries NotifyLog m blk
notifyLog)
  case Map.lookupGE cursor entries of
    Maybe (NotifyCursor, Notification blk)
Nothing -> STM m (Notification blk, NotifyCursor)
forall a. STM m a
forall (m :: * -> *) a. MonadSTM m => STM m a
retry
    Just (NotifyCursor
at, Notification blk
n) -> (Notification blk, NotifyCursor)
-> STM m (Notification blk, NotifyCursor)
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Notification blk
n, NotifyCursor -> NotifyCursor
succCursor NotifyCursor
at)

-- | Prune every notification no longer worth relaying at the given time.
--
-- Leaves the notifications untouched if there is none, so as not to wake every
-- server reading them.
pruneNotifyLog :: IOLike m => NotifyLog m blk -> RelativeTime -> STM m ()
pruneNotifyLog :: forall (m :: * -> *) blk.
IOLike m =>
NotifyLog m blk -> RelativeTime -> STM m ()
pruneNotifyLog NotifyLog m blk
notifyLog RelativeTime
now = do
  StrictTVar m RelativeTime
-> (RelativeTime -> RelativeTime) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar (NotifyLog m blk -> StrictTVar m RelativeTime
forall (m :: * -> *) blk.
NotifyLog m blk -> StrictTVar m RelativeTime
notifyPrunedAt NotifyLog m blk
notifyLog) (RelativeTime -> RelativeTime -> RelativeTime
forall a. Ord a => a -> a -> a
max RelativeTime
now)
  (expired, unexpired) <- (RelativeTime -> Bool)
-> Map RelativeTime [NotifyCursor]
-> (Map RelativeTime [NotifyCursor],
    Map RelativeTime [NotifyCursor])
forall k a. (k -> Bool) -> Map k a -> (Map k a, Map k a)
Map.spanAntitone (RelativeTime -> RelativeTime -> Bool
forall a. Ord a => a -> a -> Bool
< RelativeTime
now) (Map RelativeTime [NotifyCursor]
 -> (Map RelativeTime [NotifyCursor],
     Map RelativeTime [NotifyCursor]))
-> STM m (Map RelativeTime [NotifyCursor])
-> STM
     m
     (Map RelativeTime [NotifyCursor], Map RelativeTime [NotifyCursor])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTVar m (Map RelativeTime [NotifyCursor])
-> STM m (Map RelativeTime [NotifyCursor])
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar (NotifyLog m blk -> StrictTVar m (Map RelativeTime [NotifyCursor])
forall (m :: * -> *) blk.
NotifyLog m blk -> StrictTVar m (Map RelativeTime [NotifyCursor])
notifyExpiry NotifyLog m blk
notifyLog)
  unless (Map.null expired) $ do
    writeTVar (notifyExpiry notifyLog) unexpired
    modifyTVar (notifyEntries notifyLog) $
      (`Map.withoutKeys` Set.fromList (concat (Map.elems expired)))

succCursor :: NotifyCursor -> NotifyCursor
succCursor :: NotifyCursor -> NotifyCursor
succCursor (MkNotifyCursor Word64
n) = Word64 -> NotifyCursor
MkNotifyCursor (Word64
n Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ Word64
1)

-- | Prune the log every 'notifyLogPruneCadence'.
notifyLogPruner :: IOLike m => SystemTime m -> NotifyLog m blk -> m Void
notifyLogPruner :: forall (m :: * -> *) blk.
IOLike m =>
SystemTime m -> NotifyLog m blk -> m Void
notifyLogPruner SystemTime m
systemTime NotifyLog m blk
notifyLog = m () -> m Void
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (m () -> m Void) -> m () -> m Void
forall a b. (a -> b) -> a -> b
$ do
  now <- SystemTime m -> m RelativeTime
forall (m :: * -> *). SystemTime m -> m RelativeTime
systemTimeCurrent SystemTime m
systemTime
  atomically $ pruneNotifyLog notifyLog now
  threadDelay (nominalDelay notifyLogPruneCadence)

-- | How often the log is pruned, so how long after it stops being worth
-- relaying the node may still relay a notification. It must stay far below the
-- grace a downstream peer gives before it rejects one.
notifyLogPruneCadence :: NominalDiffTime
notifyLogPruneCadence :: NominalDiffTime
notifyLogPruneCadence = NominalDiffTime
1