{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications #-}

-- | Handlers for the UTxO RPC @SyncService@ - synchronising chain data
-- (fetching blocks, dumping history, following the tip).
module Cardano.Rpc.Server.Internal.UtxoRpc.Sync
  ( fetchBlockMethod
  , followTipMethod
  , followTipStream
  , readTipMethod
  )
where

import Cardano.Api
import Cardano.Api.Consensus qualified as Consensus
import Cardano.Rpc.Proto.Api.UtxoRpc.Sync qualified as U5c
import Cardano.Rpc.Server.Internal.Error
import Cardano.Rpc.Server.Internal.Monad
import Cardano.Rpc.Server.Internal.Tracing ()
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.Block (mkAnyChainBlock)
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.ChainPoint
  ( chainPointToBlockRef
  , mkTipBlockRef
  , tipHeaderPoint
  )
import Cardano.Rpc.Server.NodeKernelAccess

import Cardano.Ledger.BaseTypes qualified as L

import RIO

import Data.ByteString qualified as BS
import Data.ProtoLens (defMessage)
import Data.Sequence qualified as Seq
import Data.Time.Clock (UTCTime)
import GHC.Stack (withFrozenCallStack)
import Network.GRPC.Spec
  ( GrpcError (GrpcInternal, GrpcInvalidArgument, GrpcNotFound)
  , NextElem (NextElem)
  , Proto
  )

-- | Handle the @FetchBlock@ SyncService RPC method.
-- Fetches a block from ChainDB by slot and header hash.
-- Byron-era transactions carry no fee: Byron fees are implicit (inputs minus
-- outputs) and computing them needs UTxO lookups this handler does not do.
-- Returns @NOT_FOUND@ if the requested block is missing.
-- Returns @INVALID_ARGUMENT@ if the block reference has an invalid hash.
fetchBlockMethod
  :: MonadRpc e m
  => Proto U5c.FetchBlockRequest
  -- ^ Request containing a block reference (slot + hash)
  -> m (Proto U5c.FetchBlockResponse)
  -- ^ Response containing the fetched block with raw CBOR and cardano header
fetchBlockMethod :: forall e (m :: * -> *).
MonadRpc e m =>
Proto FetchBlockRequest -> m (Proto FetchBlockResponse)
fetchBlockMethod Proto FetchBlockRequest
request = do
  nodeKernelAccess@NodeKernelAccess{systemStart, readEraHistory} <- m NodeKernelAccess
forall e (m :: * -> *). MonadRpc e m => m NodeKernelAccess
grabNodeKernelAccess
  (slot, headerHash) <- blockRefToPoint (request ^. U5c.ref)
  let throwNotFound =
        GrpcError -> Text -> m (ByteString, BlockInMode)
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage GrpcError
GrpcNotFound (Text -> m (ByteString, BlockInMode))
-> Text -> m (ByteString, BlockInMode)
forall a b. (a -> b) -> a -> b
$
          Text
"block not found at slot " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Word64 -> Text
forall a. Show a => a -> Text
tshow (SlotNo -> Word64
unSlotNo SlotNo
slot)
  (rawBytes, blockInMode) <-
    fetchBlock nodeKernelAccess slot headerHash >>= maybe throwNotFound pure
  timestamp <- slotTimestampOrThrow systemStart readEraHistory slot
  pure $ defMessage & U5c.block .~ mkAnyChainBlock rawBytes blockInMode timestamp

-- | Handle the @ReadTip@ SyncService RPC method.
-- Reads the current chain tip from ChainDB and returns it as slot, block
-- header hash, block height and slot timestamp.
-- When the chain is at origin, the tip field is left unset.
readTipMethod
  :: MonadRpc e m
  => Proto U5c.ReadTipRequest
  -> m (Proto U5c.ReadTipResponse)
readTipMethod :: forall e (m :: * -> *).
MonadRpc e m =>
Proto ReadTipRequest -> m (Proto ReadTipResponse)
readTipMethod Proto ReadTipRequest
_request = do
  NodeKernelAccess{chainDb, systemStart, readEraHistory} <- m NodeKernelAccess
forall e (m :: * -> *). MonadRpc e m => m NodeKernelAccess
grabNodeKernelAccess
  tip <- readTipBlockRef chainDb (slotTimestampOrThrow systemStart readEraHistory)
  pure $ defMessage & U5c.maybe'tip .~ tip

-- | Handle the @FollowTip@ SyncService RPC method: stream fully parsed
-- blocks as the chain advances.
--
-- Where the stream starts: at the first of the request's intersection
-- points found on the chain, in client preference order. A block ref with
-- an empty hash means origin. An empty intersect list means the current
-- tip.
--
-- What the client receives: first a @reset@ announcing the start point,
-- then an @apply@ per adopted block. A rollback becomes @undo@ actions
-- carrying the rolled-back blocks, re-fetched from ChainDB and streamed
-- newest first. When the blocks can no longer be re-fetched, because
-- garbage collection won the race against the client, the rollback
-- becomes a @reset@ carrying the rollback point's @BlockRef@ instead,
-- slot and hash only, like ChainSync's @MsgRollBackward@. The tracked
-- window is sized to the node's security parameter /k/, so no rollback
-- consensus can produce falls outside it (see
-- 'Cardano.Rpc.Server.NodeKernelAccess.Type.NodeKernelAccess').
-- Every response also carries the current chain tip.
--
-- Errors: @INVALID_ARGUMENT@ if an intersection block ref has an invalid
-- hash, @NOT_FOUND@ if none of the intersection points are on the chain.
--
-- Runs until the client disconnects or the stream is otherwise closed;
-- 'withFollower' closes the follower on every exit path.
followTipMethod
  :: MonadRpc e m
  => Proto U5c.FollowTipRequest
  -- ^ Request containing optional intersection points (slot + hash)
  -> (NextElem (Proto U5c.FollowTipResponse) -> IO ())
  -- ^ Callback used to send each streamed response
  -> m ()
followTipMethod :: forall e (m :: * -> *).
MonadRpc e m =>
Proto FollowTipRequest
-> (NextElem (Proto FollowTipResponse) -> IO ()) -> m ()
followTipMethod Proto FollowTipRequest
request NextElem (Proto FollowTipResponse) -> IO ()
send = do
  nodeKernelAccess@NodeKernelAccess{chainDb, systemStart, readEraHistory, securityParam} <-
    m NodeKernelAccess
forall e (m :: * -> *). MonadRpc e m => m NodeKernelAccess
grabNodeKernelAccess
  requestedPoints <- traverse blockRefToIntersectPoint (request ^. U5c.intersect)
  withFollower nodeKernelAccess $ \ChainFollower
follower -> do
    -- an empty intersect list follows from the current tip; resolving it
    -- reaches into ChainDB directly (there is no 'ChainFollower' operation
    -- for "the current tip point"), so this step stays here rather than
    -- moving into 'followTipStream', which only takes an already-resolved,
    -- non-empty point list
    let slotTimestamp :: SlotNo -> m UTCTime
slotTimestamp = SystemStart -> m EraHistory -> SlotNo -> m UTCTime
forall (m :: * -> *).
MonadIO m =>
SystemStart -> m EraHistory -> SlotNo -> m UTCTime
slotTimestampOrThrow SystemStart
systemStart m EraHistory
forall (m :: * -> *). MonadIO m => m EraHistory
readEraHistory
    startPoints <-
      if [ChainPoint] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [ChainPoint]
requestedPoints
        then do
          tipHeader <- IO (Maybe (Header (CardanoBlock StandardCrypto)))
-> m (Maybe (Header (CardanoBlock StandardCrypto)))
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (Maybe (Header (CardanoBlock StandardCrypto)))
 -> m (Maybe (Header (CardanoBlock StandardCrypto))))
-> IO (Maybe (Header (CardanoBlock StandardCrypto)))
-> m (Maybe (Header (CardanoBlock StandardCrypto)))
forall a b. (a -> b) -> a -> b
$ ChainDB IO (CardanoBlock StandardCrypto)
-> IO (Maybe (Header (CardanoBlock StandardCrypto)))
forall (m :: * -> *) blk. ChainDB m blk -> m (Maybe (Header blk))
Consensus.getTipHeader ChainDB IO (CardanoBlock StandardCrypto)
chainDb
          pure [maybe ChainPointAtGenesis tipHeaderPoint tipHeader]
        else [ChainPoint] -> m [ChainPoint]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [ChainPoint]
requestedPoints
    followTipStream
      follower
      (readTipBlockRef chainDb slotTimestamp)
      slotTimestamp
      (fetchBlockByChainPoint nodeKernelAccess)
      (fromIntegral . L.unNonZero $ Consensus.maxRollbacks securityParam)
      send
      startPoints

-- | Convert an intersection @BlockRef@ to a 'ChainPoint'. A block ref with
-- an empty hash denotes origin, so clients can append it to the intersect
-- list as a catch-all: origin is on every chain, which makes the
-- intersection infallible.
-- Throws @INVALID_ARGUMENT@ if a non-empty hash is malformed.
blockRefToIntersectPoint
  :: MonadRpc e m
  => Proto U5c.BlockRef
  -> m ChainPoint
blockRefToIntersectPoint :: forall e (m :: * -> *).
MonadRpc e m =>
Proto BlockRef -> m ChainPoint
blockRefToIntersectPoint Proto BlockRef
blockRef
  | ByteString -> Bool
BS.null (Proto BlockRef
blockRef Proto BlockRef
-> Getting ByteString (Proto BlockRef) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto BlockRef) ByteString
forall (f :: * -> *) s a.
(Functor f, HasField s "hash" a) =>
LensLike' f s a
U5c.hash) = ChainPoint -> m ChainPoint
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ChainPoint
ChainPointAtGenesis
  | Bool
otherwise = (SlotNo -> Hash BlockHeader -> ChainPoint)
-> (SlotNo, Hash BlockHeader) -> ChainPoint
forall a b c. (a -> b -> c) -> (a, b) -> c
uncurry SlotNo -> Hash BlockHeader -> ChainPoint
ChainPoint ((SlotNo, Hash BlockHeader) -> ChainPoint)
-> m (SlotNo, Hash BlockHeader) -> m ChainPoint
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Proto BlockRef -> m (SlotNo, Hash BlockHeader)
forall e (m :: * -> *).
MonadRpc e m =>
Proto BlockRef -> m (SlotNo, Hash BlockHeader)
blockRefToPoint Proto BlockRef
blockRef

-- | Convert a @BlockRef@ into its slot and deserialised block header hash.
-- Throws @INVALID_ARGUMENT@ if the hash is malformed.
blockRefToPoint
  :: MonadRpc e m
  => Proto U5c.BlockRef
  -> m (SlotNo, Hash BlockHeader)
blockRefToPoint :: forall e (m :: * -> *).
MonadRpc e m =>
Proto BlockRef -> m (SlotNo, Hash BlockHeader)
blockRefToPoint Proto BlockRef
blockRef = do
  let slot :: SlotNo
slot = Word64 -> SlotNo
SlotNo (Word64 -> SlotNo) -> Word64 -> SlotNo
forall a b. (a -> b) -> a -> b
$ Proto BlockRef
blockRef Proto BlockRef -> Getting Word64 (Proto BlockRef) Word64 -> Word64
forall s a. s -> Getting a s a -> a
^. Getting Word64 (Proto BlockRef) Word64
forall (f :: * -> *) s a.
(Functor f, HasField s "slot" a) =>
LensLike' f s a
U5c.slot
      hashBytes :: ByteString
hashBytes = Proto BlockRef
blockRef Proto BlockRef
-> Getting ByteString (Proto BlockRef) ByteString -> ByteString
forall s a. s -> Getting a s a -> a
^. Getting ByteString (Proto BlockRef) ByteString
forall (f :: * -> *) s a.
(Functor f, HasField s "hash" a) =>
LensLike' f s a
U5c.hash
      throwInvalidHash :: m (Hash BlockHeader)
throwInvalidHash =
        GrpcError -> Text -> m (Hash BlockHeader)
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage GrpcError
GrpcInvalidArgument (Text -> m (Hash BlockHeader)) -> Text -> m (Hash BlockHeader)
forall a b. (a -> b) -> a -> b
$
          Text
"invalid block header hash (" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
tshow (ByteString -> Int
BS.length ByteString
hashBytes) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" bytes)"
  headerHash <-
    AsType (Hash BlockHeader)
-> ByteString -> Either SerialiseAsRawBytesError (Hash BlockHeader)
forall a.
SerialiseAsRawBytes a =>
AsType a -> ByteString -> Either SerialiseAsRawBytesError a
deserialiseFromRawBytes (Proxy (Hash BlockHeader) -> AsType (Hash BlockHeader)
forall t. HasTypeProxy t => Proxy t -> AsType t
proxyToAsType (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @(Hash BlockHeader))) ByteString
hashBytes
      Either SerialiseAsRawBytesError (Hash BlockHeader)
-> (Either SerialiseAsRawBytesError (Hash BlockHeader)
    -> m (Hash BlockHeader))
-> m (Hash BlockHeader)
forall a b. a -> (a -> b) -> b
& (SerialiseAsRawBytesError -> m (Hash BlockHeader))
-> (Hash BlockHeader -> m (Hash BlockHeader))
-> Either SerialiseAsRawBytesError (Hash BlockHeader)
-> m (Hash BlockHeader)
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (m (Hash BlockHeader)
-> SerialiseAsRawBytesError -> m (Hash BlockHeader)
forall a b. a -> b -> a
const m (Hash BlockHeader)
throwInvalidHash) Hash BlockHeader -> m (Hash BlockHeader)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
  pure (slot, headerHash)

-- | Adapt 'fetchBlock', which takes a slot and a hash, to the point-based
-- re-fetch parameter of 'followTipStream', which uses it to rebuild @undo@
-- payloads.
--
-- The genesis arm returns 'Nothing' only to keep the function total; it
-- cannot actually be reached. 'followTipStream' re-fetches only points
-- from its applied-points window, and every entry there comes from a
-- decoded block's header, which always has a real slot and hash (see
-- 'TrackedPoints').
fetchBlockByChainPoint
  :: MonadIO m
  => NodeKernelAccess
  -> ChainPoint
  -> m (Maybe (ByteString, BlockInMode))
fetchBlockByChainPoint :: forall (m :: * -> *).
MonadIO m =>
NodeKernelAccess
-> ChainPoint -> m (Maybe (ByteString, BlockInMode))
fetchBlockByChainPoint NodeKernelAccess
_nodeKernelAccess ChainPoint
ChainPointAtGenesis = Maybe (ByteString, BlockInMode)
-> m (Maybe (ByteString, BlockInMode))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (ByteString, BlockInMode)
forall a. Maybe a
Nothing
fetchBlockByChainPoint NodeKernelAccess
nodeKernelAccess (ChainPoint SlotNo
slot Hash BlockHeader
headerHash) =
  NodeKernelAccess
-> SlotNo
-> Hash BlockHeader
-> m (Maybe (ByteString, BlockInMode))
forall (m :: * -> *).
MonadIO m =>
NodeKernelAccess
-> SlotNo
-> Hash BlockHeader
-> m (Maybe (ByteString, BlockInMode))
fetchBlock NodeKernelAccess
nodeKernelAccess SlotNo
slot Hash BlockHeader
headerHash

-- | The applied points 'followTipStream' tracks for undo re-fetch, newest
-- first.
--
-- The entries are raw @(slot, header hash)@ pairs, not 'ChainPoint's.
-- Every entry comes from 'getBlockHeader' on a decoded 'ChainApply'
-- payload, so it always has a real slot and hash. The pair type turns
-- "an applied point is never genesis" from a convention into a fact of
-- the type: a @Seq ChainPoint@ would admit 'ChainPointAtGenesis' even
-- though it could never legitimately occur here.
type TrackedPoints = Seq.Seq (SlotNo, Hash BlockHeader)

-- | The @FollowTip@ streaming loop. Finds the intersection with the given
-- points, then streams chain changes as they arrive.
--
-- The collaborators are plain arguments rather than a 'NodeKernelAccess'
-- so tests can drive the loop with a scripted follower and stubbed
-- capabilities, no live ChainDB required (see
-- @Test.Cardano.Rpc.FollowTipStream@, the first unit coverage of this
-- loop). 'MonadIO' is enough for all of them, including the gRPC error
-- path ('throwGrpcErrorWithMessage').
--
-- How a rollback is delivered, by case:
--
-- 1. The target is within the tracked window, meaning it is one of the
--    last @trackingCap@ applied points or the window floor (the stream's
--    start point, on which a rollback undoes everything tracked). Each
--    rolled-back block is re-fetched by point and emitted as @undo@,
--    newest first.
-- 2. A re-fetch misses mid-undo because garbage collection won the race:
--    a single @reset@ at the rollback target. Partial undo followed by
--    reset is coherent because @reset@ is absolute positioning.
-- 3. The target is outside the window, either deeper than the cap or the
--    initial rollback-to-intersection when nothing is tracked yet: a
--    single @reset@, as in 2.
--
-- Every emitted message, apply or undo or reset, carries the current tip.
--
-- Throws @NOT_FOUND@ if none of the intersection points are on the chain.
-- Runs until the client disconnects or the stream is otherwise closed;
-- follower cleanup is the caller's responsibility (see 'withFollower').
followTipStream
  :: forall m
   . HasCallStack
  => MonadIO m
  => ChainFollower
  -- ^ Follower to stream changes from
  -> m (Maybe (Proto U5c.BlockRef))
  -- ^ Read the current chain tip, projected into a @BlockRef@
  -> (SlotNo -> m UTCTime)
  -- ^ Convert a slot to its wall-clock timestamp
  -> (ChainPoint -> m (Maybe (ByteString, BlockInMode)))
  -- ^ Re-fetch a block by point, to reconstruct @undo@ payloads on
  -- rollback. 'Nothing' means the block is no longer available (e.g. the
  -- VolatileDB has garbage-collected it past the immutable tip); the loop
  -- falls back to @reset@ in that case.
  -> Int
  -- ^ How many applied points to track for undo re-fetch. In production
  -- this is the node's security parameter /k/
  -- ('Cardano.Rpc.Server.NodeKernelAccess.Type.securityParam'). Consensus
  -- never rolls back more than /k/ blocks, so tracking /k/ points covers
  -- every rollback the protocol can produce, on any network. An entry
  -- costs roughly 40 bytes, so the window costs about @40 * k@ bytes per
  -- stream: around 86 KB on mainnet, where /k/ = 2160. A rollback deeper
  -- than the cap degrades to @reset@.
  -> (NextElem (Proto U5c.FollowTipResponse) -> IO ())
  -- ^ Callback used to send each streamed response
  -> [ChainPoint]
  -- ^ Resolved, non-empty intersection points, in client preference order
  -> m ()
followTipStream :: forall (m :: * -> *).
(HasCallStack, MonadIO m) =>
ChainFollower
-> m (Maybe (Proto BlockRef))
-> (SlotNo -> m UTCTime)
-> (ChainPoint -> m (Maybe (ByteString, BlockInMode)))
-> Int
-> (NextElem (Proto FollowTipResponse) -> IO ())
-> [ChainPoint]
-> m ()
followTipStream ChainFollower{forall (m :: * -> *). MonadIO m => m ChainChange
nextChange :: forall (m :: * -> *). MonadIO m => m ChainChange
nextChange :: ChainFollower -> forall (m :: * -> *). MonadIO m => m ChainChange
nextChange, forall (m :: * -> *).
MonadIO m =>
[ChainPoint] -> m (Maybe ChainPoint)
findIntersect :: forall (m :: * -> *).
MonadIO m =>
[ChainPoint] -> m (Maybe ChainPoint)
findIntersect :: ChainFollower
-> forall (m :: * -> *).
   MonadIO m =>
   [ChainPoint] -> m (Maybe ChainPoint)
findIntersect} m (Maybe (Proto BlockRef))
readTip SlotNo -> m UTCTime
slotTimestamp ChainPoint -> m (Maybe (ByteString, BlockInMode))
fetchBlockByPoint Int
trackingCap NextElem (Proto FollowTipResponse) -> IO ()
send [ChainPoint]
startPoints =
  -- freezes the caller's call stack (e.g. 'followTipMethod's) so the
  -- @NOT_FOUND@ thrown below, and any exception raised by the collaborator
  -- actions threaded through the loop, points at the real call site rather
  -- than somewhere inside this loop. This is what the 'HasCallStack'
  -- constraint above is for.
  (HasCallStack => m ()) -> m ()
forall a. HasCallStack => (HasCallStack => a) -> a
withFrozenCallStack ((HasCallStack => m ()) -> m ()) -> (HasCallStack => m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ do
    resolvedIntersection <- [ChainPoint] -> m (Maybe ChainPoint)
forall (m :: * -> *).
MonadIO m =>
[ChainPoint] -> m (Maybe ChainPoint)
findIntersect [ChainPoint]
startPoints
    startPoint <- case resolvedIntersection of
      Maybe ChainPoint
Nothing ->
        GrpcError -> Text -> m ChainPoint
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage GrpcError
GrpcNotFound (Text -> m ChainPoint) -> Text -> m ChainPoint
forall a b. (a -> b) -> a -> b
$
          Text
"no intersection found: none of the "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
tshow ([ChainPoint] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [ChainPoint]
startPoints)
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" intersect points are on the chain"
      Just ChainPoint
point -> ChainPoint -> m ChainPoint
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ChainPoint
point
    -- after a successful 'findIntersect' the follower's next instruction is
    -- a 'RollBack' to the intersection - the loop reports it as the initial
    -- 'reset' announcing where the stream starts (the "nothing tracked yet"
    -- branch of 'handleRollback' below, since the intersection is also the
    -- initial window floor and nothing has been applied yet)
    go startPoint Seq.empty
 where
  sendMessage :: Proto FollowTipResponse -> m ()
sendMessage Proto FollowTipResponse
action = do
    tip <- m (Maybe (Proto BlockRef))
readTip
    liftIO . send . NextElem $ action & U5c.maybe'tip .~ tip

  go :: ChainPoint -> TrackedPoints -> m ()
  go :: ChainPoint -> TrackedPoints -> m ()
go ChainPoint
floorPoint TrackedPoints
tracked = do
    change <- m ChainChange
forall (m :: * -> *). MonadIO m => m ChainChange
nextChange
    (floorPoint', tracked') <- case change of
      ChainApply (ByteString
rawBytes, blockInMode :: BlockInMode
blockInMode@(BlockInMode CardanoEra era
_ Block era
block)) -> do
        let BlockHeader SlotNo
slot Hash BlockHeader
headerHash BlockNo
_ = Block era -> BlockHeader
forall era. Block era -> BlockHeader
getBlockHeader Block era
block
        timestamp <- SlotNo -> m UTCTime
slotTimestamp SlotNo
slot
        sendMessage $ defMessage & U5c.apply .~ mkAnyChainBlock rawBytes blockInMode timestamp
        pure (floorPoint, trackApplied trackingCap (slot, headerHash) tracked)
      ChainRollBack ChainPoint
point -> ChainPoint
-> ChainPoint -> TrackedPoints -> m (ChainPoint, TrackedPoints)
handleRollback ChainPoint
point ChainPoint
floorPoint TrackedPoints
tracked
    go floorPoint' tracked'

  -- \| Dispatch a rollback to undo or reset. The window floor starts as
  -- the stream's start point and only ever moves forward, to a rollback
  -- target that fell outside the window (the 'Nothing' case below). A
  -- later rollback landing on the new floor can then be served as undo
  -- instead of degrading to reset a second time.
  handleRollback :: ChainPoint -> ChainPoint -> TrackedPoints -> m (ChainPoint, TrackedPoints)
  handleRollback :: ChainPoint
-> ChainPoint -> TrackedPoints -> m (ChainPoint, TrackedPoints)
handleRollback ChainPoint
point ChainPoint
floorPoint TrackedPoints
tracked =
    case ChainPoint
-> ChainPoint
-> TrackedPoints
-> Maybe (TrackedPoints, TrackedPoints)
windowSplit ChainPoint
point ChainPoint
floorPoint TrackedPoints
tracked of
      Maybe (TrackedPoints, TrackedPoints)
Nothing -> do
        -- deeper than the cap, below the stream's start, or the initial
        -- rollback-to-intersection with nothing tracked yet: today's
        -- unchanged fallback, which also preserves the first-message-reset
        -- invariant. The target becomes the new floor.
        Proto FollowTipResponse -> m ()
sendMessage (Proto FollowTipResponse -> m ())
-> Proto FollowTipResponse -> m ()
forall a b. (a -> b) -> a -> b
$ Proto FollowTipResponse
forall msg. Message msg => msg
defMessage Proto FollowTipResponse
-> (Proto FollowTipResponse -> Proto FollowTipResponse)
-> Proto FollowTipResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto FollowTipResponse) (Proto BlockRef)
forall (f :: * -> *) s a.
(Functor f, HasField s "reset" a) =>
LensLike' f s a
U5c.reset LensLike' Identity (Proto FollowTipResponse) (Proto BlockRef)
-> Proto BlockRef
-> Proto FollowTipResponse
-> Proto FollowTipResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ChainPoint -> Proto BlockRef
chainPointToBlockRef ChainPoint
point
        (ChainPoint, TrackedPoints) -> m (ChainPoint, TrackedPoints)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ChainPoint
point, TrackedPoints
forall a. Seq a
Seq.empty)
      Just (TrackedPoints
undone, TrackedPoints
kept)
        | TrackedPoints -> Bool
forall a. Seq a -> Bool
Seq.null TrackedPoints
undone ->
            -- nothing newer than the target is tracked (the stream-opening
            -- rollback, or a rollback that is a no-op against the current
            -- position): reset communicates the position, with nothing to
            -- undo
            Proto FollowTipResponse -> m ()
sendMessage (Proto FollowTipResponse
forall msg. Message msg => msg
defMessage Proto FollowTipResponse
-> (Proto FollowTipResponse -> Proto FollowTipResponse)
-> Proto FollowTipResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto FollowTipResponse) (Proto BlockRef)
forall (f :: * -> *) s a.
(Functor f, HasField s "reset" a) =>
LensLike' f s a
U5c.reset LensLike' Identity (Proto FollowTipResponse) (Proto BlockRef)
-> Proto BlockRef
-> Proto FollowTipResponse
-> Proto FollowTipResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ChainPoint -> Proto BlockRef
chainPointToBlockRef ChainPoint
point)
              m ()
-> (ChainPoint, TrackedPoints) -> m (ChainPoint, TrackedPoints)
forall (f :: * -> *) a b. Functor f => f a -> b -> f b
$> (ChainPoint
floorPoint, TrackedPoints
kept)
        | Bool
otherwise -> do
            kept' <- ChainPoint -> TrackedPoints -> TrackedPoints -> m TrackedPoints
undoNewestFirst ChainPoint
point TrackedPoints
kept TrackedPoints
undone
            pure (floorPoint, kept')

  -- \| Split the tracked points at the rollback target. 'Nothing' means
  -- the target is outside the window: not the floor and not a tracked
  -- point. Otherwise the first half is the points strictly newer than the
  -- target, to be undone newest first, and the second half is what
  -- survives: the target's own entry, if it is tracked, and everything
  -- older.
  windowSplit :: ChainPoint -> ChainPoint -> TrackedPoints -> Maybe (TrackedPoints, TrackedPoints)
  windowSplit :: ChainPoint
-> ChainPoint
-> TrackedPoints
-> Maybe (TrackedPoints, TrackedPoints)
windowSplit ChainPoint
point ChainPoint
floorPoint TrackedPoints
tracked = case ChainPoint
point of
    ChainPoint SlotNo
slot Hash BlockHeader
headerHash
      | Just Int
i <- ((SlotNo, Hash BlockHeader) -> Bool) -> TrackedPoints -> Maybe Int
forall a. (a -> Bool) -> Seq a -> Maybe Int
Seq.findIndexL ((SlotNo, Hash BlockHeader) -> (SlotNo, Hash BlockHeader) -> Bool
forall a. Eq a => a -> a -> Bool
== (SlotNo
slot, Hash BlockHeader
headerHash)) TrackedPoints
tracked ->
          (TrackedPoints, TrackedPoints)
-> Maybe (TrackedPoints, TrackedPoints)
forall a. a -> Maybe a
Just (Int -> TrackedPoints -> TrackedPoints
forall a. Int -> Seq a -> Seq a
Seq.take Int
i TrackedPoints
tracked, Int -> TrackedPoints -> TrackedPoints
forall a. Int -> Seq a -> Seq a
Seq.drop Int
i TrackedPoints
tracked)
    ChainPoint
_
      | ChainPoint
point ChainPoint -> ChainPoint -> Bool
forall a. Eq a => a -> a -> Bool
== ChainPoint
floorPoint -> (TrackedPoints, TrackedPoints)
-> Maybe (TrackedPoints, TrackedPoints)
forall a. a -> Maybe a
Just (TrackedPoints
tracked, TrackedPoints
forall a. Seq a
Seq.empty)
      | Bool
otherwise -> Maybe (TrackedPoints, TrackedPoints)
forall a. Maybe a
Nothing

  -- \| Re-fetch and emit @undo@ for each pending point, newest first.
  -- Stops at the first fetch miss (garbage collection won the race) and
  -- sends a single absolute @reset@ at the rollback point instead.
  undoNewestFirst :: ChainPoint -> TrackedPoints -> TrackedPoints -> m TrackedPoints
  undoNewestFirst :: ChainPoint -> TrackedPoints -> TrackedPoints -> m TrackedPoints
undoNewestFirst ChainPoint
point TrackedPoints
kept = TrackedPoints -> m TrackedPoints
loop
   where
    loop :: TrackedPoints -> m TrackedPoints
loop TrackedPoints
pending = case TrackedPoints -> ViewL (SlotNo, Hash BlockHeader)
forall a. Seq a -> ViewL a
Seq.viewl TrackedPoints
pending of
      ViewL (SlotNo, Hash BlockHeader)
Seq.EmptyL -> TrackedPoints -> m TrackedPoints
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure TrackedPoints
kept
      (SlotNo
slot, Hash BlockHeader
headerHash) Seq.:< TrackedPoints
rest -> do
        fetched <- ChainPoint -> m (Maybe (ByteString, BlockInMode))
fetchBlockByPoint (SlotNo -> Hash BlockHeader -> ChainPoint
ChainPoint SlotNo
slot Hash BlockHeader
headerHash)
        case fetched of
          Just (ByteString
rawBytes, BlockInMode
blockInMode) -> do
            timestamp <- SlotNo -> m UTCTime
slotTimestamp SlotNo
slot
            sendMessage $ defMessage & U5c.undo .~ mkAnyChainBlock rawBytes blockInMode timestamp
            loop rest
          Maybe (ByteString, BlockInMode)
Nothing ->
            Proto FollowTipResponse -> m ()
sendMessage (Proto FollowTipResponse
forall msg. Message msg => msg
defMessage Proto FollowTipResponse
-> (Proto FollowTipResponse -> Proto FollowTipResponse)
-> Proto FollowTipResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto FollowTipResponse) (Proto BlockRef)
forall (f :: * -> *) s a.
(Functor f, HasField s "reset" a) =>
LensLike' f s a
U5c.reset LensLike' Identity (Proto FollowTipResponse) (Proto BlockRef)
-> Proto BlockRef
-> Proto FollowTipResponse
-> Proto FollowTipResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ ChainPoint -> Proto BlockRef
chainPointToBlockRef ChainPoint
point) m () -> TrackedPoints -> m TrackedPoints
forall (f :: * -> *) a b. Functor f => f a -> b -> f b
$> TrackedPoints
kept

  trackApplied :: Int -> (SlotNo, Hash BlockHeader) -> TrackedPoints -> TrackedPoints
  trackApplied :: Int -> (SlotNo, Hash BlockHeader) -> TrackedPoints -> TrackedPoints
trackApplied Int
cap (SlotNo, Hash BlockHeader)
entry TrackedPoints
tracked = Int -> TrackedPoints -> TrackedPoints
forall a. Int -> Seq a -> Seq a
Seq.take Int
cap ((SlotNo, Hash BlockHeader)
entry (SlotNo, Hash BlockHeader) -> TrackedPoints -> TrackedPoints
forall a. a -> Seq a -> Seq a
Seq.<| TrackedPoints
tracked)

-- | Read the current chain tip and project it into a @BlockRef@ via
-- 'mkTipBlockRef', or 'Nothing' at origin.
readTipBlockRef
  :: MonadIO m
  => Consensus.ChainDB IO (Consensus.CardanoBlock Consensus.StandardCrypto)
  -> (SlotNo -> m UTCTime)
  -- ^ Convert a slot to its wall-clock timestamp
  -> m (Maybe (Proto U5c.BlockRef))
readTipBlockRef :: forall (m :: * -> *).
MonadIO m =>
ChainDB IO (CardanoBlock StandardCrypto)
-> (SlotNo -> m UTCTime) -> m (Maybe (Proto BlockRef))
readTipBlockRef ChainDB IO (CardanoBlock StandardCrypto)
chainDb SlotNo -> m UTCTime
slotTimestamp = do
  tipHeader <- IO (Maybe (Header (CardanoBlock StandardCrypto)))
-> m (Maybe (Header (CardanoBlock StandardCrypto)))
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (Maybe (Header (CardanoBlock StandardCrypto)))
 -> m (Maybe (Header (CardanoBlock StandardCrypto))))
-> IO (Maybe (Header (CardanoBlock StandardCrypto)))
-> m (Maybe (Header (CardanoBlock StandardCrypto)))
forall a b. (a -> b) -> a -> b
$ ChainDB IO (CardanoBlock StandardCrypto)
-> IO (Maybe (Header (CardanoBlock StandardCrypto)))
forall (m :: * -> *) blk. ChainDB m blk -> m (Maybe (Header blk))
Consensus.getTipHeader ChainDB IO (CardanoBlock StandardCrypto)
chainDb
  forM tipHeader $ \Header (CardanoBlock StandardCrypto)
header ->
    Header (CardanoBlock StandardCrypto) -> UTCTime -> Proto BlockRef
mkTipBlockRef Header (CardanoBlock StandardCrypto)
header (UTCTime -> Proto BlockRef) -> m UTCTime -> m (Proto BlockRef)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> SlotNo -> m UTCTime
slotTimestamp (Header (CardanoBlock StandardCrypto) -> SlotNo
forall b. HasHeader b => b -> SlotNo
Consensus.blockSlot Header (CardanoBlock StandardCrypto)
header)

-- | Convert a slot to its wall-clock timestamp.
-- Throws @INTERNAL@ when the slot is past the era history horizon.
slotTimestampOrThrow
  :: MonadIO m
  => SystemStart
  -> m EraHistory
  -- ^ Read current era history from the ledger state
  -> SlotNo
  -> m UTCTime
slotTimestampOrThrow :: forall (m :: * -> *).
MonadIO m =>
SystemStart -> m EraHistory -> SlotNo -> m UTCTime
slotTimestampOrThrow SystemStart
systemStart m EraHistory
readEraHistory SlotNo
slot = do
  eraHistory <- m EraHistory
readEraHistory
  slotToUTCTime systemStart eraHistory slot
    & either (const throwPastHorizon) pure
 where
  throwPastHorizon :: m UTCTime
throwPastHorizon =
    GrpcError -> Text -> m UTCTime
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage GrpcError
GrpcInternal (Text -> m UTCTime) -> Text -> m UTCTime
forall a b. (a -> b) -> a -> b
$
      Text
"cannot convert slot "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Word64 -> Text
forall a. Show a => a -> Text
tshow (SlotNo -> Word64
unSlotNo SlotNo
slot)
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" to timestamp: the slot is past the era history horizon;"
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" check that the requested slot is correct and that the node is fully in sync"