{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE RankNTypes #-}

-- | Chain followers over the node's ChainDB: the follower handle and the
-- brackets that open one. Re-exported from "Cardano.Rpc.Server.NodeKernelAccess".
module Cardano.Rpc.Server.NodeKernelAccess.Internal.Follower
  ( ChainChange (..)
  , ChainFollower (..)
  , withFollower
  , withBlockFollower
  , positionFollowerAtCurrentTip
  , readChainTipPoint
  )
where

import Cardano.Api
import Cardano.Api.Consensus qualified as Consensus
import Cardano.Rpc.Server.NodeKernelAccess.Internal.Type qualified as Type

import Ouroboros.Consensus.Storage.ChainDB.API qualified as Consensus (getTipPoint)

import RIO (MonadUnliftIO, atomically, bracket, withRunInIO)

import Control.Monad.STM (STM)
import Data.ByteString (ByteString)
import Data.ByteString.Lazy qualified as BSL

-- | A handle to a running chain follower. The payload is selectable because
-- consensus reads each requested block component separately.
data ChainFollower a = ChainFollower
  { forall a.
ChainFollower a
-> forall (m :: * -> *). MonadIO m => m (ChainChange a)
nextChange :: forall m. MonadIO m => m (ChainChange a)
  -- ^ Block until the next chain update is available.
  , forall a.
ChainFollower a
-> forall (m :: * -> *). MonadIO m => m (Maybe (ChainChange a))
tryNextChange :: forall m. MonadIO m => m (Maybe (ChainChange a))
  -- ^ The next chain update, or 'Nothing' if there is none yet. Does not
  -- block, but may read the block from disk.
  , forall a.
ChainFollower a
-> forall (m :: * -> *).
   MonadIO m =>
   [ChainPoint] -> m (Maybe ChainPoint)
findIntersect :: forall m. MonadIO m => [ChainPoint] -> m (Maybe ChainPoint)
  -- ^ Move the follower to the first of the given points found on the
  -- current chain, returning that point, or 'Nothing' if none of them are
  -- on the chain.
  }

-- | A single instruction produced by a chain follower.
--
-- 'ChainApply' carries the per-block payload the follower was opened for
-- (see 'withFollower' and 'withBlockFollower').
-- Consensus rollbacks are point-only: 'ChainRollBack' never carries the
-- blocks being rolled back, only the point to roll back to.
data ChainChange a
  = ChainApply a
  | ChainRollBack ChainPoint

-- | Run an action with a 'ChainFollower' tracking the selected chain, whose
-- 'ChainApply' payload is the raw CBOR block bytes together with the same
-- block parsed into its era context - exactly the pair 'fetchBlock' returns.
--
-- The follower and the resource registry backing it are closed on every
-- exit path, including exceptions. The follower itself runs in 'IO',
-- because the ChainDB handle is monomorphic, so the bracket runs there and
-- the action is unlifted into it.
--
-- Creating a follower is cheap: a few in-memory STM operations, nothing
-- proportional to chain length. The costs are steady-state instead. A
-- caught-up follower receives an O(1) notification per adopted block. A
-- follower catching up streams blocks from the ImmutableDB, paying a disk
-- read and a deserialisation per block, with file handles owned by the
-- registry. The node already runs one such follower per connected N2C
-- ChainSync client, so one follower per stream scales the same way.
withFollower
  :: MonadUnliftIO m
  => Type.NodeKernelAccess
  -> (ChainFollower (ByteString, BlockInMode) -> m a)
  -> m a
withFollower :: forall (m :: * -> *) a.
MonadUnliftIO m =>
NodeKernelAccess
-> (ChainFollower (ByteString, BlockInMode) -> m a) -> m a
withFollower =
  BlockComponent
  (CardanoBlock StandardCrypto) (ByteString, BlockInMode)
-> NodeKernelAccess
-> (ChainFollower (ByteString, BlockInMode) -> m a)
-> m a
forall (m :: * -> *) b a.
MonadUnliftIO m =>
BlockComponent (CardanoBlock StandardCrypto) b
-> NodeKernelAccess -> (ChainFollower b -> m a) -> m a
withFollowerReading (BlockComponent
   (CardanoBlock StandardCrypto) (ByteString, BlockInMode)
 -> NodeKernelAccess
 -> (ChainFollower (ByteString, BlockInMode) -> m a)
 -> m a)
-> BlockComponent
     (CardanoBlock StandardCrypto) (ByteString, BlockInMode)
-> NodeKernelAccess
-> (ChainFollower (ByteString, BlockInMode) -> m a)
-> m a
forall a b. (a -> b) -> a -> b
$
    (,) (ByteString -> BlockInMode -> (ByteString, BlockInMode))
-> BlockComponent (CardanoBlock StandardCrypto) ByteString
-> BlockComponent
     (CardanoBlock StandardCrypto)
     (BlockInMode -> (ByteString, BlockInMode))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (ByteString -> ByteString)
-> BlockComponent (CardanoBlock StandardCrypto) ByteString
-> BlockComponent (CardanoBlock StandardCrypto) ByteString
forall a b.
(a -> b)
-> BlockComponent (CardanoBlock StandardCrypto) a
-> BlockComponent (CardanoBlock StandardCrypto) b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ByteString -> ByteString
BSL.toStrict BlockComponent (CardanoBlock StandardCrypto) ByteString
forall blk. BlockComponent blk ByteString
Consensus.GetRawBlock BlockComponent
  (CardanoBlock StandardCrypto)
  (BlockInMode -> (ByteString, BlockInMode))
-> BlockComponent (CardanoBlock StandardCrypto) BlockInMode
-> BlockComponent
     (CardanoBlock StandardCrypto) (ByteString, BlockInMode)
forall a b.
BlockComponent (CardanoBlock StandardCrypto) (a -> b)
-> BlockComponent (CardanoBlock StandardCrypto) a
-> BlockComponent (CardanoBlock StandardCrypto) b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (CardanoBlock StandardCrypto -> BlockInMode)
-> BlockComponent
     (CardanoBlock StandardCrypto) (CardanoBlock StandardCrypto)
-> BlockComponent (CardanoBlock StandardCrypto) BlockInMode
forall a b.
(a -> b)
-> BlockComponent (CardanoBlock StandardCrypto) a
-> BlockComponent (CardanoBlock StandardCrypto) b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap CardanoBlock StandardCrypto -> BlockInMode
forall block.
(CardanoBlock StandardCrypto ~ block) =>
block -> BlockInMode
fromConsensusBlock BlockComponent
  (CardanoBlock StandardCrypto) (CardanoBlock StandardCrypto)
forall blk. BlockComponent blk blk
Consensus.GetBlock

-- | 'withFollower' with the parsed block alone as the 'ChainApply' payload,
-- reading each block once instead of twice.
withBlockFollower
  :: MonadUnliftIO m
  => Type.NodeKernelAccess
  -> (ChainFollower BlockInMode -> m a)
  -> m a
withBlockFollower :: forall (m :: * -> *) a.
MonadUnliftIO m =>
NodeKernelAccess -> (ChainFollower BlockInMode -> m a) -> m a
withBlockFollower = BlockComponent (CardanoBlock StandardCrypto) BlockInMode
-> NodeKernelAccess -> (ChainFollower BlockInMode -> m a) -> m a
forall (m :: * -> *) b a.
MonadUnliftIO m =>
BlockComponent (CardanoBlock StandardCrypto) b
-> NodeKernelAccess -> (ChainFollower b -> m a) -> m a
withFollowerReading (BlockComponent (CardanoBlock StandardCrypto) BlockInMode
 -> NodeKernelAccess -> (ChainFollower BlockInMode -> m a) -> m a)
-> BlockComponent (CardanoBlock StandardCrypto) BlockInMode
-> NodeKernelAccess
-> (ChainFollower BlockInMode -> m a)
-> m a
forall a b. (a -> b) -> a -> b
$ CardanoBlock StandardCrypto -> BlockInMode
forall block.
(CardanoBlock StandardCrypto ~ block) =>
block -> BlockInMode
fromConsensusBlock (CardanoBlock StandardCrypto -> BlockInMode)
-> BlockComponent
     (CardanoBlock StandardCrypto) (CardanoBlock StandardCrypto)
-> BlockComponent (CardanoBlock StandardCrypto) BlockInMode
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> BlockComponent
  (CardanoBlock StandardCrypto) (CardanoBlock StandardCrypto)
forall blk. BlockComponent blk blk
Consensus.GetBlock

-- | Run an action with a follower that reads the given block component.
withFollowerReading
  :: MonadUnliftIO m
  => Consensus.BlockComponent (Consensus.CardanoBlock Consensus.StandardCrypto) b
  -> Type.NodeKernelAccess
  -> (ChainFollower b -> m a)
  -> m a
withFollowerReading :: forall (m :: * -> *) b a.
MonadUnliftIO m =>
BlockComponent (CardanoBlock StandardCrypto) b
-> NodeKernelAccess -> (ChainFollower b -> m a) -> m a
withFollowerReading BlockComponent (CardanoBlock StandardCrypto) b
component Type.NodeKernelAccess{chainDb :: NodeKernelAccess -> ChainDB IO (CardanoBlock StandardCrypto)
Type.chainDb = ChainDB IO (CardanoBlock StandardCrypto)
chainDb} ChainFollower b -> m a
action =
  ((forall a. m a -> IO a) -> IO a) -> m a
forall b. ((forall a. m a -> IO a) -> IO b) -> m b
forall (m :: * -> *) b.
MonadUnliftIO m =>
((forall a. m a -> IO a) -> IO b) -> m b
withRunInIO (((forall a. m a -> IO a) -> IO a) -> m a)
-> ((forall a. m a -> IO a) -> IO a) -> m a
forall a b. (a -> b) -> a -> b
$ \forall a. m a -> IO a
runInIO ->
    (ResourceRegistry IO -> IO a) -> IO a
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, MonadEvaluate m,
 HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
Consensus.withRegistry ((ResourceRegistry IO -> IO a) -> IO a)
-> (ResourceRegistry IO -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \ResourceRegistry IO
registry ->
      IO (Follower IO (CardanoBlock StandardCrypto) b)
-> (Follower IO (CardanoBlock StandardCrypto) b -> IO ())
-> (Follower IO (CardanoBlock StandardCrypto) b -> IO a)
-> IO a
forall (m :: * -> *) a b c.
MonadUnliftIO m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket
        (ChainDB IO (CardanoBlock StandardCrypto)
-> forall b.
   ResourceRegistry IO
   -> ChainType
   -> BlockComponent (CardanoBlock StandardCrypto) b
   -> IO (Follower IO (CardanoBlock StandardCrypto) b)
forall (m :: * -> *) blk.
ChainDB m blk
-> forall b.
   ResourceRegistry m
   -> ChainType -> BlockComponent blk b -> m (Follower m blk b)
Consensus.newFollower ChainDB IO (CardanoBlock StandardCrypto)
chainDb ResourceRegistry IO
registry ChainType
Consensus.SelectedChain BlockComponent (CardanoBlock StandardCrypto) b
component)
        Follower IO (CardanoBlock StandardCrypto) b -> IO ()
forall (m :: * -> *) blk a. Follower m blk a -> m ()
Consensus.followerClose
        (m a -> IO a
forall a. m a -> IO a
runInIO (m a -> IO a)
-> (Follower IO (CardanoBlock StandardCrypto) b -> m a)
-> Follower IO (CardanoBlock StandardCrypto) b
-> IO a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChainFollower b -> m a
action (ChainFollower b -> m a)
-> (Follower IO (CardanoBlock StandardCrypto) b -> ChainFollower b)
-> Follower IO (CardanoBlock StandardCrypto) b
-> m a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Follower IO (CardanoBlock StandardCrypto) b -> ChainFollower b
forall b.
Follower IO (CardanoBlock StandardCrypto) b -> ChainFollower b
toChainFollower)
 where
  toChainFollower
    :: Consensus.Follower IO (Consensus.CardanoBlock Consensus.StandardCrypto) b
    -> ChainFollower b
  toChainFollower :: forall b.
Follower IO (CardanoBlock StandardCrypto) b -> ChainFollower b
toChainFollower Follower IO (CardanoBlock StandardCrypto) b
follower =
    ChainFollower
      { nextChange :: forall (m :: * -> *). MonadIO m => m (ChainChange b)
nextChange = IO (ChainChange b) -> m (ChainChange b)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (ChainChange b) -> m (ChainChange b))
-> IO (ChainChange b) -> m (ChainChange b)
forall a b. (a -> b) -> a -> b
$ ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b
forall b.
ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b
toChainChange (ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b)
-> IO (ChainUpdate (CardanoBlock StandardCrypto) b)
-> IO (ChainChange b)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Follower IO (CardanoBlock StandardCrypto) b
-> IO (ChainUpdate (CardanoBlock StandardCrypto) b)
forall (m :: * -> *) blk a.
Follower m blk a -> m (ChainUpdate blk a)
Consensus.followerInstructionBlocking Follower IO (CardanoBlock StandardCrypto) b
follower
      , tryNextChange :: forall (m :: * -> *). MonadIO m => m (Maybe (ChainChange b))
tryNextChange = IO (Maybe (ChainChange b)) -> m (Maybe (ChainChange b))
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (Maybe (ChainChange b)) -> m (Maybe (ChainChange b)))
-> IO (Maybe (ChainChange b)) -> m (Maybe (ChainChange b))
forall a b. (a -> b) -> a -> b
$ (ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b)
-> Maybe (ChainUpdate (CardanoBlock StandardCrypto) b)
-> Maybe (ChainChange b)
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b
forall b.
ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b
toChainChange (Maybe (ChainUpdate (CardanoBlock StandardCrypto) b)
 -> Maybe (ChainChange b))
-> IO (Maybe (ChainUpdate (CardanoBlock StandardCrypto) b))
-> IO (Maybe (ChainChange b))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Follower IO (CardanoBlock StandardCrypto) b
-> IO (Maybe (ChainUpdate (CardanoBlock StandardCrypto) b))
forall (m :: * -> *) blk a.
Follower m blk a -> m (Maybe (ChainUpdate blk a))
Consensus.followerInstruction Follower IO (CardanoBlock StandardCrypto) b
follower
      , findIntersect :: forall (m :: * -> *).
MonadIO m =>
[ChainPoint] -> m (Maybe ChainPoint)
findIntersect = \[ChainPoint]
points ->
          IO (Maybe ChainPoint) -> m (Maybe ChainPoint)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (Maybe ChainPoint) -> m (Maybe ChainPoint))
-> IO (Maybe ChainPoint) -> m (Maybe ChainPoint)
forall a b. (a -> b) -> a -> b
$
            (Point (CardanoBlock StandardCrypto) -> ChainPoint)
-> Maybe (Point (CardanoBlock StandardCrypto)) -> Maybe ChainPoint
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Point (CardanoBlock StandardCrypto) -> ChainPoint
forall block (xs :: [*]).
(HeaderHash block ~ OneEraHash xs) =>
Point block -> ChainPoint
fromConsensusPointHF
              (Maybe (Point (CardanoBlock StandardCrypto)) -> Maybe ChainPoint)
-> IO (Maybe (Point (CardanoBlock StandardCrypto)))
-> IO (Maybe ChainPoint)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Follower IO (CardanoBlock StandardCrypto) b
-> [Point (CardanoBlock StandardCrypto)]
-> IO (Maybe (Point (CardanoBlock StandardCrypto)))
forall (m :: * -> *) blk a.
Follower m blk a -> [Point blk] -> m (Maybe (Point blk))
Consensus.followerForward Follower IO (CardanoBlock StandardCrypto) b
follower ((ChainPoint -> Point (CardanoBlock StandardCrypto))
-> [ChainPoint] -> [Point (CardanoBlock StandardCrypto)]
forall a b. (a -> b) -> [a] -> [b]
map ChainPoint -> Point (CardanoBlock StandardCrypto)
forall block (xs :: [*]).
(HeaderHash block ~ OneEraHash xs) =>
ChainPoint -> Point block
toConsensusPointHF [ChainPoint]
points)
      }

  toChainChange
    :: Consensus.ChainUpdate (Consensus.CardanoBlock Consensus.StandardCrypto) b
    -> ChainChange b
  toChainChange :: forall b.
ChainUpdate (CardanoBlock StandardCrypto) b -> ChainChange b
toChainChange = \case
    Consensus.AddBlock b
payload -> b -> ChainChange b
forall a. a -> ChainChange a
ChainApply b
payload
    Consensus.RollBack Point (CardanoBlock StandardCrypto)
point -> ChainPoint -> ChainChange b
forall a. ChainPoint -> ChainChange a
ChainRollBack (Point (CardanoBlock StandardCrypto) -> ChainPoint
forall block (xs :: [*]).
(HeaderHash block ~ OneEraHash xs) =>
Point block -> ChainPoint
fromConsensusPointHF Point (CardanoBlock StandardCrypto)
point)

-- | Position a freshly opened chain follower at the current tip (origin, at
-- genesis), so it reports only blocks applied from now on, and return that
-- point. A fork switch between the tip read and the intersection takes the
-- old tip off the chain; then read the new tip and try again, rather than
-- fall back to origin and replay the whole chain on the first drain.
positionFollowerAtCurrentTip
  :: MonadIO m
  => Type.NodeKernelAccess
  -> ChainFollower a
  -> m ChainPoint
positionFollowerAtCurrentTip :: forall (m :: * -> *) a.
MonadIO m =>
NodeKernelAccess -> ChainFollower a -> m ChainPoint
positionFollowerAtCurrentTip NodeKernelAccess
nodeKernelAccess ChainFollower{forall (m :: * -> *).
MonadIO m =>
[ChainPoint] -> m (Maybe ChainPoint)
findIntersect :: forall a.
ChainFollower a
-> forall (m :: * -> *).
   MonadIO m =>
   [ChainPoint] -> m (Maybe ChainPoint)
findIntersect :: forall (m :: * -> *).
MonadIO m =>
[ChainPoint] -> m (Maybe ChainPoint)
findIntersect} = m ChainPoint
go
 where
  go :: m ChainPoint
go = do
    tipPoint <- STM ChainPoint -> m ChainPoint
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM ChainPoint -> m ChainPoint) -> STM ChainPoint -> m ChainPoint
forall a b. (a -> b) -> a -> b
$ NodeKernelAccess -> STM ChainPoint
readChainTipPoint NodeKernelAccess
nodeKernelAccess
    findIntersect [tipPoint] >>= maybe go pure

-- | Read the current chain tip point from ChainDB, or 'ChainPointAtGenesis'
-- at origin. An STM read, so a caller can wait for the tip to move.
readChainTipPoint :: Type.NodeKernelAccess -> STM ChainPoint
readChainTipPoint :: NodeKernelAccess -> STM ChainPoint
readChainTipPoint Type.NodeKernelAccess{chainDb :: NodeKernelAccess -> ChainDB IO (CardanoBlock StandardCrypto)
Type.chainDb = ChainDB IO (CardanoBlock StandardCrypto)
chainDb} =
  Point (CardanoBlock StandardCrypto) -> ChainPoint
forall block (xs :: [*]).
(HeaderHash block ~ OneEraHash xs) =>
Point block -> ChainPoint
fromConsensusPointHF (Point (CardanoBlock StandardCrypto) -> ChainPoint)
-> STM (Point (CardanoBlock StandardCrypto)) -> STM ChainPoint
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ChainDB IO (CardanoBlock StandardCrypto)
-> STM IO (Point (CardanoBlock StandardCrypto))
forall (m :: * -> *) blk. ChainDB m blk -> STM m (Point blk)
Consensus.getTipPoint ChainDB IO (CardanoBlock StandardCrypto)
chainDb