Skip to content

Commit fe4b3cc

Browse files
carbolymercarbolymer
committed
phase 4- wip
1 parent c5a909d commit fe4b3cc

8 files changed

Lines changed: 232 additions & 95 deletions

File tree

‎.changes/20260722_cardano_rpc_follow_tip.yml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
project: cardano-rpc
2+
pr: 0
3+
kind:
4+
- feature
5+
description: |
6+
Implement the FollowTip SyncService method: streams fully parsed blocks as the chain advances, starting from the first intersection point found on the chain (an empty-hash block ref denotes origin). An empty intersect list follows from the current tip; an unmatched intersect list fails with NOT_FOUND. Rollbacks are delivered as reset actions.

‎cardano-rpc/README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ It implements [UTxO RPC](https://utxorpc.org/introduction) protobuf communicatio
3535
|--------|--------|
3636
| [FetchBlock](https://utxorpc.org/sync/spec/#fetchblockrequest) | ✅ Supported |
3737
| [DumpHistory](https://utxorpc.org/sync/spec/#dumphistoryrequest) | ⬜ Not supported |
38-
| [FollowTip](https://utxorpc.org/sync/spec/#followtiprequest) | ⬜ Not supported |
38+
| [FollowTip](https://utxorpc.org/sync/spec/#followtiprequest) | ✅ Supported |
3939
| [ReadTip](https://utxorpc.org/sync/spec/#readtiprequest) | ✅ Supported |
4040

4141
### [WatchService](https://utxorpc.org/watch/spec/)

‎cardano-rpc/cardano-rpc.cabal‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ library
6666
Cardano.Rpc.Server.Internal.UtxoRpc.Sync
6767
Cardano.Rpc.Server.Internal.UtxoRpc.Type
6868
Cardano.Rpc.Server.Internal.UtxoRpc.Type.BigInt
69+
Cardano.Rpc.Server.Internal.UtxoRpc.Type.Block
6970
Cardano.Rpc.Server.Internal.UtxoRpc.Type.Byron
7071
Cardano.Rpc.Server.Internal.UtxoRpc.Type.Certificate
7172
Cardano.Rpc.Server.Internal.UtxoRpc.Type.ChainPoint

‎cardano-rpc/src/Cardano/Rpc/Server.hs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ methodsSyncRpc
8484
methodsSyncRpc =
8585
Method (mkNonStreaming $ const unimplemented) -- dumpHistory
8686
. Method (mkNonStreaming $ wrapInSpan TraceRpcFetchBlockSpan . fetchBlockMethod)
87-
. Method (mkServerStreaming $ \_ _ -> unimplemented) -- followTip
87+
. Method (mkServerStreaming $ \req -> wrapInSpan TraceRpcFollowTipSpan . followTipMethod req)
8888
. Method (mkNonStreaming $ wrapInSpan TraceRpcReadTipSpan . readTipMethod)
8989
$ NoMoreMethods
9090
where

‎cardano-rpc/src/Cardano/Rpc/Server/Internal/Tracing.hs‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -101,12 +101,12 @@ data TraceRpcSync
101101
TraceRpcFetchBlockSpan TraceSpanEvent
102102
| -- | ReadTip span
103103
TraceRpcReadTipSpan TraceSpanEvent
104+
| -- | FollowTip span
105+
TraceRpcFollowTipSpan TraceSpanEvent
104106
| -- | Requested block was not found
105107
TraceRpcFetchBlockNotFound SlotNo
106108
| -- | Node kernel access is not yet available
107109
TraceRpcNodeKernelAccessUnavailable
108-
| -- | Ledger forker error
109-
TraceRpcForkerError String
110110
deriving Show
111111

112112
instance Pretty TraceRpcSync where
@@ -115,9 +115,10 @@ instance Pretty TraceRpcSync where
115115
TraceRpcFetchBlockSpan (SpanEnd _) -> "Finished FetchBlock method"
116116
TraceRpcReadTipSpan (SpanBegin _) -> "Started ReadTip method"
117117
TraceRpcReadTipSpan (SpanEnd _) -> "Finished ReadTip method"
118+
TraceRpcFollowTipSpan (SpanBegin _) -> "Started FollowTip method"
119+
TraceRpcFollowTipSpan (SpanEnd _) -> "Finished FollowTip method"
118120
TraceRpcFetchBlockNotFound slot -> "Block not found at slot " <> pshow slot
119121
TraceRpcNodeKernelAccessUnavailable -> "Node kernel access not yet initialised"
120-
TraceRpcForkerError e -> "Ledger forker error: " <> pretty e
121122

122123
instance Error TraceRpcSync where
123124
prettyError = pretty

‎cardano-rpc/src/Cardano/Rpc/Server/Internal/UtxoRpc/Sync.hs‎

Lines changed: 121 additions & 88 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
-- (fetching blocks, dumping history, following the tip).
99
module Cardano.Rpc.Server.Internal.UtxoRpc.Sync
1010
( fetchBlockMethod
11+
, followTipMethod
1112
, readTipMethod
1213
)
1314
where
@@ -18,18 +19,24 @@ import Cardano.Rpc.Proto.Api.UtxoRpc.Sync qualified as U5c
1819
import Cardano.Rpc.Server.Internal.Error
1920
import Cardano.Rpc.Server.Internal.Monad
2021
import Cardano.Rpc.Server.Internal.Tracing ()
21-
import Cardano.Rpc.Server.Internal.UtxoRpc.Type (anyEraTxConstraints, txToUtxoRpcTx)
22-
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.Byron (byronBlockTxs)
22+
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.Block (mkAnyChainBlock)
23+
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.ChainPoint
24+
( chainPointToBlockRef
25+
, mkTipBlockRef
26+
, tipHeaderPoint
27+
)
2328
import Cardano.Rpc.Server.NodeKernelAccess
2429

2530
import RIO
2631

2732
import Data.ByteString qualified as BS
28-
import Data.ByteString.Short qualified as SBS
2933
import Data.ProtoLens (defMessage)
3034
import Data.Time.Clock (UTCTime)
31-
import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
32-
import Network.GRPC.Spec (GrpcError (GrpcInternal, GrpcInvalidArgument, GrpcNotFound), Proto)
35+
import Network.GRPC.Spec
36+
( GrpcError (GrpcInvalidArgument, GrpcNotFound)
37+
, NextElem (NextElem)
38+
, Proto
39+
)
3340

3441
-- | Handle the @FetchBlock@ SyncService RPC method.
3542
-- Fetches a block from ChainDB by slot and header hash.
@@ -44,31 +51,14 @@ fetchBlockMethod
4451
-> m (Proto U5c.FetchBlockResponse)
4552
-- ^ Response containing the fetched block with raw CBOR and cardano header
4653
fetchBlockMethod request = do
47-
nodeKernelAccess@NodeKernelAccess{systemStart, readEraHistory} <- grabNodeKernelAccess
48-
let blockRef = request ^. U5c.ref
49-
slot = SlotNo $ blockRef ^. U5c.slot
50-
hashBytes = blockRef ^. U5c.hash
51-
throwInvalidHash =
52-
throwGrpcErrorWithMessage GrpcInvalidArgument $
53-
"invalid block header hash (" <> tshow (BS.length hashBytes) <> " bytes)"
54-
throwNotFound =
54+
nodeKernelAccess <- grabNodeKernelAccess
55+
(slot, headerHash) <- blockRefToPoint (request ^. U5c.ref)
56+
let throwNotFound =
5557
throwGrpcErrorWithMessage GrpcNotFound $
5658
"block not found at slot " <> tshow (unSlotNo slot)
57-
throwPastHorizon =
58-
throwGrpcErrorWithMessage GrpcInternal $
59-
"cannot convert slot "
60-
<> tshow (unSlotNo slot)
61-
<> " to timestamp: the slot is past the era history horizon;"
62-
<> " check that the requested slot is correct and that the node is fully in sync"
63-
headerHash <-
64-
deserialiseFromRawBytes (proxyToAsType (Proxy @(Hash BlockHeader))) hashBytes
65-
& either (const throwInvalidHash) pure
6659
(rawBytes, blockInMode) <-
6760
fetchBlock nodeKernelAccess slot headerHash >>= maybe throwNotFound pure
68-
eraHistory <- readEraHistory
69-
timestamp <-
70-
slotToUTCTime systemStart eraHistory slot
71-
& either (const throwPastHorizon) pure
61+
timestamp <- slotTimestampOrThrow nodeKernelAccess slot
7262
pure $ defMessage & U5c.block .~ mkAnyChainBlock rawBytes blockInMode timestamp
7363

7464
-- | Handle the @ReadTip@ SyncService RPC method.
@@ -80,69 +70,112 @@ readTipMethod
8070
=> Proto U5c.ReadTipRequest
8171
-> m (Proto U5c.ReadTipResponse)
8272
readTipMethod _request = do
83-
NodeKernelAccess{chainDb, systemStart, readEraHistory} <- grabNodeKernelAccess
84-
tipHeader <- liftIO $ Consensus.getTipHeader chainDb
85-
tip <- forM tipHeader $ \header -> do
86-
let slot = Consensus.blockSlot header
87-
throwPastHorizon =
88-
throwGrpcErrorWithMessage GrpcInternal $
89-
"cannot convert tip slot "
90-
<> tshow (unSlotNo slot)
91-
<> " to timestamp: the slot is past the era history horizon"
92-
eraHistory <- readEraHistory
93-
timestamp <-
94-
slotToUTCTime systemStart eraHistory slot
95-
& either (const throwPastHorizon) pure
96-
pure $ mkTipBlockRef header timestamp
73+
nodeKernelAccess <- grabNodeKernelAccess
74+
tip <- readTipBlockRef nodeKernelAccess
9775
pure $ defMessage & U5c.maybe'tip .~ tip
9876

99-
-- | Assemble the @AnyChainBlock@ proto message: raw CBOR bytes, the cardano
100-
-- header (slot, hash, height - derived from the block itself) and the parsed
101-
-- transactions (all eras), plus the given slot timestamp.
102-
mkAnyChainBlock
103-
:: ByteString
104-
-> BlockInMode
105-
-> UTCTime
106-
-- ^ Slot wall-clock time; encoded as milliseconds since the Unix epoch
107-
-> Proto U5c.AnyChainBlock
108-
mkAnyChainBlock rawBytes (BlockInMode _ block) timestamp =
109-
let BlockHeader slot headerHash (BlockNo height) = getBlockHeader block
110-
-- Byron transactions are not representable as cardano-api's 'Tx era',
111-
-- so they are converted straight from the Byron ledger types
112-
txs = case block of
113-
ByronBlock consensusBlock ->
114-
byronBlockTxs (byronBlockRaw consensusBlock)
115-
ShelleyBlock sbe _ ->
116-
anyEraTxConstraints sbe $
117-
getBlockTxs block <&> \(ShelleyTx _ ledgerTx) -> txToUtxoRpcTx ledgerTx
118-
blockHeader =
119-
defMessage
120-
& U5c.slot .~ unSlotNo slot
121-
& U5c.hash .~ serialiseToRawBytes headerHash
122-
& U5c.height .~ height
123-
in defMessage
124-
& U5c.nativeBytes .~ rawBytes
125-
& U5c.cardano . U5c.header .~ blockHeader
126-
& U5c.cardano . U5c.body . U5c.tx .~ txs
127-
& U5c.cardano . U5c.timestamp .~ utcTimeToMs timestamp
77+
-- | Handle the @FollowTip@ SyncService RPC method.
78+
-- Streams fully parsed blocks as the chain advances, starting from the
79+
-- first of the request's intersection points found on the chain (client
80+
-- preference order). An intersection block ref with an empty hash denotes
81+
-- origin; when the intersect list is empty, the stream follows from the
82+
-- current tip.
83+
-- The first streamed message is always a @reset@ announcing where the
84+
-- stream starts. Later rollbacks are delivered the same way, as @reset@
85+
-- actions carrying the rollback point's @BlockRef@ (slot and hash only).
86+
-- Every response also carries the current chain tip.
87+
-- Returns @INVALID_ARGUMENT@ if an intersection block reference has an
88+
-- invalid hash and @NOT_FOUND@ if none of the intersection points are on
89+
-- the chain.
90+
-- Runs until the client disconnects or the stream is otherwise closed; the
91+
-- follower is closed on every exit path by 'withFollower'.
92+
followTipMethod
93+
:: MonadRpc e m
94+
=> Proto U5c.FollowTipRequest
95+
-- ^ Request containing optional intersection points (slot + hash)
96+
-> (NextElem (Proto U5c.FollowTipResponse) -> IO ())
97+
-- ^ Callback used to send each streamed response
98+
-> m ()
99+
followTipMethod request send = do
100+
nodeKernelAccess@NodeKernelAccess{chainDb} <- grabNodeKernelAccess
101+
requestedPoints <- traverse blockRefToIntersectPoint (request ^. U5c.intersect)
102+
withFollower nodeKernelAccess $ \ChainFollower{nextChange, findIntersect} -> do
103+
-- an empty intersect list follows from the current tip
104+
startPoints <-
105+
if null requestedPoints
106+
then do
107+
tipHeader <- liftIO $ Consensus.getTipHeader chainDb
108+
pure [maybe ChainPointAtGenesis tipHeaderPoint tipHeader]
109+
else pure requestedPoints
110+
intersection <- findIntersect startPoints
111+
when (isNothing intersection) $
112+
throwGrpcErrorWithMessage GrpcNotFound $
113+
"no intersection found: none of the "
114+
<> tshow (length startPoints)
115+
<> " intersect points are on the chain"
116+
-- after a successful 'findIntersect' the follower's next instruction is
117+
-- a 'RollBack' to the intersection - the loop reports it as the initial
118+
-- 'reset' announcing where the stream starts
119+
forever $ do
120+
change <- nextChange
121+
action <- case change of
122+
ChainApply (rawBytes, blockInMode@(BlockInMode _ block)) -> do
123+
let BlockHeader slot _ _ = getBlockHeader block
124+
timestamp <- slotTimestampOrThrow nodeKernelAccess slot
125+
pure $ defMessage & U5c.apply .~ mkAnyChainBlock rawBytes blockInMode timestamp
126+
ChainRollBack point ->
127+
pure $ defMessage & U5c.reset .~ chainPointToBlockRef point
128+
tip <- readTipBlockRef nodeKernelAccess
129+
liftIO . send . NextElem $ action & U5c.maybe'tip .~ tip
128130

129-
-- | Project a ChainDB header and a slot timestamp into a @BlockRef@: slot,
130-
-- header hash, block height and timestamp.
131-
mkTipBlockRef
132-
:: Consensus.Header (Consensus.CardanoBlock Consensus.StandardCrypto)
133-
-> UTCTime
134-
-- ^ Slot wall-clock time; encoded as milliseconds since the Unix epoch
135-
-> Proto U5c.BlockRef
136-
mkTipBlockRef header timestamp =
137-
let slot = Consensus.blockSlot header
138-
Consensus.OneEraHash tipHash = Consensus.blockHash header
139-
BlockNo height = Consensus.blockNo header
140-
in defMessage
141-
& U5c.slot .~ unSlotNo slot
142-
& U5c.hash .~ SBS.fromShort tipHash
143-
& U5c.height .~ height
144-
& U5c.timestamp .~ utcTimeToMs timestamp
131+
-- | Convert an intersection @BlockRef@ to a 'ChainPoint'. A block ref with
132+
-- an empty hash denotes origin, so clients can append it to the intersect
133+
-- list as a catch-all: origin is on every chain, which makes the
134+
-- intersection infallible.
135+
-- Throws @INVALID_ARGUMENT@ if a non-empty hash is malformed.
136+
blockRefToIntersectPoint
137+
:: MonadRpc e m
138+
=> Proto U5c.BlockRef
139+
-> m ChainPoint
140+
blockRefToIntersectPoint blockRef
141+
| BS.null (blockRef ^. U5c.hash) = pure ChainPointAtGenesis
142+
| otherwise = uncurry ChainPoint <$> blockRefToPoint blockRef
145143

146-
-- | Milliseconds since the Unix epoch, the timestamp encoding UTxO RPC uses.
147-
utcTimeToMs :: UTCTime -> Word64
148-
utcTimeToMs = round . (* 1000) . utcTimeToPOSIXSeconds
144+
-- | Convert a @BlockRef@ into its slot and deserialised block header hash.
145+
-- Throws @INVALID_ARGUMENT@ if the hash is malformed.
146+
blockRefToPoint
147+
:: MonadRpc e m
148+
=> Proto U5c.BlockRef
149+
-> m (SlotNo, Hash BlockHeader)
150+
blockRefToPoint blockRef = do
151+
let slot = SlotNo $ blockRef ^. U5c.slot
152+
hashBytes = blockRef ^. U5c.hash
153+
throwInvalidHash =
154+
throwGrpcErrorWithMessage GrpcInvalidArgument $
155+
"invalid block header hash (" <> tshow (BS.length hashBytes) <> " bytes)"
156+
headerHash <-
157+
deserialiseFromRawBytes (proxyToAsType (Proxy @(Hash BlockHeader))) hashBytes
158+
& either (const throwInvalidHash) pure
159+
pure (slot, headerHash)
160+
161+
-- | Read the current chain tip and project it into a @BlockRef@ via
162+
-- 'mkTipBlockRef', or 'Nothing' at origin.
163+
readTipBlockRef
164+
:: MonadRpc e m
165+
=> NodeKernelAccess
166+
-> m (Maybe (Proto U5c.BlockRef))
167+
readTipBlockRef nodeKernelAccess@NodeKernelAccess{chainDb} = do
168+
tipHeader <- liftIO $ Consensus.getTipHeader chainDb
169+
forM tipHeader $ \header ->
170+
mkTipBlockRef header <$> slotTimestampOrThrow nodeKernelAccess (Consensus.blockSlot header)
171+
172+
-- | Convert a slot to its wall-clock timestamp.
173+
-- Throws when the slot is past the era history horizon.
174+
slotTimestampOrThrow
175+
:: MonadRpc e m
176+
=> NodeKernelAccess
177+
-> SlotNo
178+
-> m UTCTime
179+
slotTimestampOrThrow NodeKernelAccess{systemStart, readEraHistory} slot = do
180+
eraHistory <- readEraHistory
181+
throwEither $ slotToUTCTime systemStart eraHistory slot

‎cardano-rpc/src/Cardano/Rpc/Server/Internal/UtxoRpc/Type/Block.hs‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
{-# LANGUAGE GADTs #-}
2+
3+
-- | Conversion of a fetched or streamed chain block to the UTxO RPC
4+
-- @AnyChainBlock@ message.
5+
module Cardano.Rpc.Server.Internal.UtxoRpc.Type.Block
6+
( mkAnyChainBlock
7+
)
8+
where
9+
10+
import Cardano.Api.Block
11+
import Cardano.Api.Consensus (byronBlockRaw)
12+
import Cardano.Api.Serialise.Raw
13+
import Cardano.Api.Tx
14+
import Cardano.Rpc.Proto.Api.UtxoRpc.Sync qualified as U5c
15+
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.Byron (byronBlockTxs)
16+
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.ChainPoint (utcTimeToMs)
17+
import Cardano.Rpc.Server.Internal.UtxoRpc.Type.Tx (anyEraTxConstraints, txToUtxoRpcTx)
18+
19+
import RIO
20+
21+
import Data.ProtoLens (defMessage)
22+
import Data.Time.Clock (UTCTime)
23+
import Network.GRPC.Spec
24+
25+
-- | Assemble the @AnyChainBlock@ proto message: raw CBOR bytes, the cardano
26+
-- header (slot, hash, height - derived from the block itself) and the parsed
27+
-- transactions (all eras), plus the given slot timestamp.
28+
mkAnyChainBlock
29+
:: ByteString
30+
-> BlockInMode
31+
-> UTCTime
32+
-- ^ Slot wall-clock time; encoded as milliseconds since the Unix epoch
33+
-> Proto U5c.AnyChainBlock
34+
mkAnyChainBlock rawBytes (BlockInMode _ block) timestamp =
35+
let BlockHeader slot headerHash (BlockNo height) = getBlockHeader block
36+
-- Byron transactions are not representable as cardano-api's 'Tx era',
37+
-- so they are converted straight from the Byron ledger types
38+
txs = case block of
39+
ByronBlock consensusBlock ->
40+
byronBlockTxs (byronBlockRaw consensusBlock)
41+
ShelleyBlock sbe _ ->
42+
anyEraTxConstraints sbe $
43+
getBlockTxs block <&> \(ShelleyTx _ ledgerTx) -> txToUtxoRpcTx ledgerTx
44+
blockHeader =
45+
defMessage
46+
& U5c.slot .~ unSlotNo slot
47+
& U5c.hash .~ serialiseToRawBytes headerHash
48+
& U5c.height .~ height
49+
in defMessage
50+
& U5c.nativeBytes .~ rawBytes
51+
& U5c.cardano . U5c.header .~ blockHeader
52+
& U5c.cardano . U5c.body . U5c.tx .~ txs
53+
& U5c.cardano . U5c.timestamp .~ utcTimeToMs timestamp

0 commit comments

Comments
 (0)