Skip to content

Commit ab05007

Browse files
eyeinskyeyeinsky
committed
Add EpochNonce indexer and a combining EpochState
1 parent 6e627b7 commit ab05007

File tree

8 files changed

+343
-133
lines changed

8 files changed

+343
-133
lines changed

‎mafoc/app/Main.hs

+8
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,9 @@ import Cardano.Streaming.Callbacks qualified as CS
1111
import Mafoc.CLI qualified as Opt
1212
import Mafoc.Core (runIndexer)
1313
import Mafoc.Indexers.BlockBasics qualified as BlockBasics
14+
import Mafoc.Indexers.EpochNonce qualified as EpochNonce
1415
import Mafoc.Indexers.EpochStakepoolSize qualified as EpochStakepoolSize
16+
import Mafoc.Indexers.EpochState qualified as EpochState
1517
import Mafoc.Indexers.MintBurn qualified as MintBurn
1618
import Mafoc.Indexers.NoOp qualified as NoOp
1719
import Mafoc.Indexers.ScriptTx qualified as ScriptTx
@@ -27,6 +29,8 @@ main = printRollbackException $ Opt.execParser cmdParserInfo >>= \case
2729
MintBurn configFromCli -> runIndexer configFromCli
2830
NoOp configFromCli -> runIndexer configFromCli
2931
EpochStakepoolSize configFromCli -> runIndexer configFromCli
32+
EpochNonce configFromCli -> runIndexer configFromCli
33+
EpochState configFromCli -> runIndexer configFromCli
3034
ScriptTx configFromCli -> runIndexer configFromCli
3135

3236
printRollbackException :: IO () -> IO ()
@@ -40,6 +44,8 @@ data Command
4044
| MintBurn MintBurn.MintBurn
4145
| NoOp NoOp.NoOp
4246
| EpochStakepoolSize EpochStakepoolSize.EpochStakepoolSize
47+
| EpochNonce EpochNonce.EpochNonce
48+
| EpochState EpochState.EpochState
4349
| ScriptTx ScriptTx.ScriptTx
4450
deriving Show
4551

@@ -55,6 +61,8 @@ cmdParser = Opt.subparser
5561
<> Opt.command "mintburn" (MintBurn <$> MintBurn.parseCli)
5662
<> Opt.command "noop" (NoOp <$> NoOp.parseCli)
5763
<> Opt.command "epochstakepoolsize" (EpochStakepoolSize <$> EpochStakepoolSize.parseCli)
64+
<> Opt.command "epochnonce" (EpochNonce <$> EpochNonce.parseCli)
65+
<> Opt.command "epochstate" (EpochState <$> EpochState.parseCli)
5866

5967
speedParserInfo :: Opt.ParserInfo Command
6068
speedParserInfo = Opt.info parser help

‎mafoc/mafoc.cabal

+3
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,9 @@ library
5050
Mafoc.Core
5151
Mafoc.Indexer.Class
5252
Mafoc.Indexers.BlockBasics
53+
Mafoc.Indexers.EpochNonce
5354
Mafoc.Indexers.EpochStakepoolSize
55+
Mafoc.Indexers.EpochState
5456
Mafoc.Indexers.MintBurn
5557
Mafoc.Indexers.NoOp
5658
Mafoc.Indexers.ScriptTx
@@ -73,6 +75,7 @@ library
7375
--------------------------
7476
build-depends:
7577
, cardano-api
78+
, cardano-ledger-shelley
7679
, iohk-monitoring
7780
, ouroboros-consensus
7881

‎mafoc/mafoc.cabal.dev

+3
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,9 @@ library
5050
Mafoc.Core
5151
Mafoc.Indexer.Class
5252
Mafoc.Indexers.BlockBasics
53+
Mafoc.Indexers.EpochNonce
5354
Mafoc.Indexers.EpochStakepoolSize
55+
Mafoc.Indexers.EpochState
5456
Mafoc.Indexers.MintBurn
5557
Mafoc.Indexers.NoOp
5658
Mafoc.Indexers.ScriptTx
@@ -75,6 +77,7 @@ library
7577
, iohk-monitoring
7678
, ouroboros-consensus
7779
, ouroboros-consensus-cardano
80+
, cardano-ledger-shelley
7881

7982
------------------------
8083
-- Non-IOG dependencies

‎mafoc/src/Mafoc/CLI.hs

+1-13
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
module Mafoc.CLI where
22

3-
import Data.ByteString.Char8 qualified as C8
43
import Data.List qualified as L
5-
import Data.Proxy (Proxy (Proxy))
64
import Data.Word (Word32)
75
import Numeric.Natural (Natural)
86
import Options.Applicative ((<|>))
@@ -12,7 +10,7 @@ import Text.Read qualified as Read
1210
import Cardano.Api qualified as C
1311
import Mafoc.Core (ConcurrencyPrimitive, DbPathAndTableName (DbPathAndTableName), Interval (Interval),
1412
LocalChainsyncConfig (LocalChainsyncConfig), LocalChainsyncConfig_, NodeConfig, NodeInfo,
15-
UpTo (CurrentTip, Infinity, SlotNo))
13+
UpTo (CurrentTip, Infinity, SlotNo), eitherParseHashBlockHeader, leftError, parseSlotNo_)
1614
import Marconi.ChainIndex.Types qualified as Marconi
1715

1816
-- * Options
@@ -125,19 +123,9 @@ mkCommonLocalChainsyncConfig commonNodeConnection_ = LocalChainsyncConfig
125123

126124
-- * String parsers
127125

128-
parseSlotNo_ :: String -> Either String C.SlotNo
129-
parseSlotNo_ str = maybe (leftError "Can't read SlotNo" str) (Right . C.SlotNo) $ Read.readMaybe str
130-
131-
leftError :: String -> String -> Either String a
132-
leftError label str = Left $ label <> ": '" <> str <> "'"
133-
134126
maybeParseHashBlockHeader :: String -> Maybe (C.Hash C.BlockHeader)
135127
maybeParseHashBlockHeader = either (const Nothing) Just . eitherParseHashBlockHeader
136128

137-
eitherParseHashBlockHeader = C.deserialiseFromRawBytesHex (C.proxyToAsType Proxy) . C8.pack
138-
139-
eitherParseHashBlockHeader_ = either (Left . show) Right . eitherParseHashBlockHeader
140-
141129
-- ** Interval
142130

143131
parseIntervalEither :: String -> Either String Interval

‎mafoc/src/Mafoc/Core.hs

+142-16
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,48 @@
1-
{-# LANGUAGE AllowAmbiguousTypes #-}
2-
{-# LANGUAGE LambdaCase #-}
3-
{-# LANGUAGE MultiWayIf #-}
4-
{-# LANGUAGE OverloadedStrings #-}
5-
{-# LANGUAGE RankNTypes #-}
6-
{-# LANGUAGE TypeFamilyDependencies #-}
1+
{-# LANGUAGE AllowAmbiguousTypes #-}
2+
{-# LANGUAGE LambdaCase #-}
3+
{-# LANGUAGE MultiWayIf #-}
4+
{-# LANGUAGE OverloadedStrings #-}
5+
{-# LANGUAGE RankNTypes #-}
6+
{-# LANGUAGE TupleSections #-}
77
module Mafoc.Core
88
( module Mafoc.Core
99
, module Mafoc.Upstream
1010
) where
1111

12+
import Control.Concurrent qualified as IO
13+
import Control.Concurrent.STM qualified as STM
14+
import Control.Concurrent.STM.TChan qualified as TChan
15+
import Control.Monad.Except (runExceptT)
1216
import Control.Monad.Trans.Class (lift)
13-
import Data.Function ((&))
14-
import Data.Maybe (fromMaybe)
17+
import Data.ByteString.Char8 qualified as C8
18+
import Data.Coerce (coerce)
19+
import Data.Function (on, (&))
20+
import Data.List qualified as L
21+
import Data.Maybe (fromMaybe, mapMaybe)
22+
import Data.Proxy (Proxy (Proxy))
1523
import Data.Text qualified as TS
1624
import Data.Time (UTCTime, diffUTCTime, getCurrentTime)
17-
import Data.Word (Word32)
25+
import Data.Word (Word32, Word64)
1826
import Database.SQLite.Simple qualified as SQL
1927
import Numeric.Natural (Natural)
28+
import Prettyprinter (Pretty (pretty), defaultLayoutOptions, layoutPretty)
29+
import Prettyprinter.Render.Text (renderStrict)
2030
import Streaming qualified as S
2131
import Streaming.Prelude qualified as S
32+
import System.Directory (listDirectory, removeFile)
2233
import System.FilePath ((</>))
23-
24-
import Control.Concurrent qualified as IO
25-
import Control.Concurrent.STM qualified as STM
26-
import Control.Concurrent.STM.TChan qualified as TChan
34+
import Text.Read qualified as Read
2735

2836
import Cardano.Api qualified as C
29-
-- import Cardano.BM.Data.Trace
3037
import Cardano.BM.Setup (withTrace)
3138
import Cardano.BM.Trace qualified as Trace
3239
import Cardano.BM.Tracing (defaultConfigStdout)
3340
import Cardano.Streaming qualified as CS
41+
import Marconi.ChainIndex.Indexers.EpochState qualified as Marconi
3442
import Marconi.ChainIndex.Indexers.MintBurn ()
3543
import Marconi.ChainIndex.Types qualified as Marconi
36-
import Prettyprinter (Pretty (pretty), defaultLayoutOptions, layoutPretty)
37-
import Prettyprinter.Render.Text (renderStrict)
44+
import Ouroboros.Consensus.Config qualified as O
45+
import Ouroboros.Consensus.Ledger.Extended qualified as O
3846

3947
import Mafoc.RollbackRingBuffer qualified as RB
4048
import Mafoc.Logging qualified as Logging
@@ -178,6 +186,16 @@ nodeFolderToSocketPath nodeFolder = nodeFolder </> "socket" </> "node.socket"
178186
nodeInfoSocketPath :: Either NodeFolder (SocketPath, a) -> SocketPath
179187
nodeInfoSocketPath nodeInfo_ = either nodeFolderToSocketPath fst nodeInfo_
180188

189+
getNodeConfigSocketPath :: LocalChainsyncConfig NodeConfig -> (NodeConfig, SocketPath)
190+
getNodeConfigSocketPath chainsyncConfigWithNodeConfig = (nodeConfig, socketPath)
191+
where
192+
nodeInfo' = nodeInfo chainsyncConfigWithNodeConfig
193+
nodeConfig = case nodeInfo' of
194+
Left nodeFolder -> nodeFolderToConfigPath nodeFolder
195+
Right (_socketPath, nodeConfig') -> nodeConfig'
196+
socketPath = nodeInfoSocketPath nodeInfo'
197+
198+
181199
-- | Resolve @LocalChainsyncConfig@ that came from e.g command line
182200
-- arguments into an "actionable" @LocalChainsyncRuntime@ runtime
183201
-- config which can be used to generate a stream of blocks.
@@ -306,6 +324,114 @@ traceInfo trace msg = Trace.logInfo trace $ renderStrict $ layoutPretty defaultL
306324
traceDebug :: Trace.Trace IO TS.Text -> String -> IO ()
307325
traceDebug trace msg = Trace.logDebug trace $ renderStrict $ layoutPretty defaultLayoutOptions $ pretty msg
308326

327+
-- * Ledger state checkpoint
328+
329+
listExtLedgerStates :: FilePath -> IO [(FilePath, SlotNoBhh)]
330+
listExtLedgerStates dirPath = L.sortBy (flip compare `on` snd) . mapMaybe parse <$> listDirectory dirPath
331+
where
332+
parse :: FilePath -> Maybe (FilePath, SlotNoBhh)
333+
parse fn = either (const Nothing) Just . fmap (fn,) $ bhhFromFileName fn
334+
335+
loadLedgerState :: FilePath -> Trace.Trace IO TS.Text -> IO (Marconi.ExtLedgerCfg_, Marconi.ExtLedgerState_, C.ChainPoint)
336+
loadLedgerState nodeConfig trace = do
337+
paths <- listExtLedgerStates "."
338+
case paths of
339+
-- A ledger state exists on disk, resume from there
340+
(fn, (slotNo, bhh)) : _ -> do
341+
cfg <- Marconi.getLedgerConfig nodeConfig
342+
let O.ExtLedgerCfg topLevelConfig = cfg
343+
extLedgerState <- Marconi.loadExtLedgerState (O.configCodec topLevelConfig) fn >>= \case
344+
Right (_, extLedgerState) -> return extLedgerState
345+
Left msg -> error $ "Error while deserialising file " <> fn <> ", error: " <> show msg
346+
let cp = C.ChainPoint slotNo bhh
347+
traceInfo trace $ "Found on-disk ledger state, resuming from: " <> show cp
348+
return (cfg, extLedgerState, cp)
349+
-- No existing ledger states, start from the beginning
350+
[] -> do
351+
(cfg, st) <- Marconi.getInitialExtLedgerState nodeConfig
352+
traceInfo trace "No on-disk ledger state found, resuming from genesis"
353+
return (cfg, st, C.ChainPointAtGenesis)
354+
355+
storeLedgerState :: Marconi.ExtLedgerCfg_ -> SlotNoBhh -> Marconi.ExtLedgerState_ -> IO ()
356+
storeLedgerState ledgerCfg slotNoBhh extLedgerState = do
357+
let O.ExtLedgerCfg topLevelConfig = ledgerCfg
358+
putStrLn $ "Write ledger state"
359+
Marconi.writeExtLedgerState (bhhToFileName slotNoBhh) (O.configCodec topLevelConfig) extLedgerState
360+
putStrLn $ "Wrote ledger state"
361+
mapM_ (removeFile . bhhToFileName) . drop 2 . map snd =<< listExtLedgerStates "."
362+
putStrLn $ "Removed other files"
363+
364+
-- | Initialization for ledger state indexers
365+
initializeLedgerState
366+
:: LocalChainsyncConfig NodeConfig
367+
-> Trace.Trace IO TS.Text
368+
-> DbPathAndTableName -- ^ Path to sqlite db and table name from cli
369+
-> (SQL.Connection -> String -> IO ()) -- ^ Function which takes a connection and a table name and creates the table.
370+
-> String
371+
-> IO ( Marconi.ExtLedgerState_, Maybe C.EpochNo
372+
, LocalChainsyncRuntime
373+
, SQL.Connection, String, Marconi.ExtLedgerCfg_)
374+
initializeLedgerState chainsyncConfig trace dbPathAndTableName sqliteInit defaultTableName' = do
375+
let (nodeConfig, socketPath) = getNodeConfigSocketPath chainsyncConfig
376+
377+
networkId <- #getNetworkId nodeConfig
378+
let localNodeConnectInfo = CS.mkLocalNodeConnectInfo networkId socketPath
379+
securityParam' <- querySecurityParam localNodeConnectInfo
380+
let (dbPath, tableName) = defaultTableName defaultTableName' dbPathAndTableName
381+
sqlCon <- sqliteOpen dbPath
382+
sqliteInit sqlCon tableName
383+
384+
(ledgerConfig, extLedgerState, startFrom) <- loadLedgerState nodeConfig trace
385+
386+
let chainsyncRuntime' = LocalChainsyncRuntime
387+
localNodeConnectInfo
388+
((interval_ chainsyncConfig) {from = startFrom})
389+
securityParam'
390+
(logging_ chainsyncConfig)
391+
(pipelineSize_ chainsyncConfig)
392+
(batchSize_ chainsyncConfig)
393+
(concurrencyPrimitive_ chainsyncConfig)
394+
395+
return ( extLedgerState, Marconi.getEpochNo extLedgerState
396+
, chainsyncRuntime'
397+
, sqlCon, tableName, ledgerConfig)
398+
399+
400+
bhhToFileName :: SlotNoBhh -> FilePath
401+
bhhToFileName (slotNo, blockHeaderHash) = L.intercalate "_"
402+
[ "ledgerState"
403+
, show slotNo'
404+
, TS.unpack (C.serialiseToRawBytesHexText blockHeaderHash)
405+
]
406+
where
407+
slotNo' = coerce slotNo :: Word64
408+
409+
bhhFromFileName :: String -> Either String SlotNoBhh
410+
bhhFromFileName str = case splitOn '_' str of
411+
_ : slotNoStr : blockHeaderHashHex : _ -> (,)
412+
<$> parseSlotNo_ slotNoStr
413+
<*> eitherParseHashBlockHeader_ blockHeaderHashHex
414+
_ -> Left "Can't parse ledger state file name, must be <slot no> _ <block header hash>"
415+
where
416+
417+
splitOn :: Eq a => a -> [a] -> [[a]]
418+
splitOn x xs = case span (/= x) xs of
419+
(prefix, _x : rest) -> prefix : recurse rest
420+
(lastChunk, []) -> [lastChunk]
421+
where
422+
recurse = splitOn x
423+
424+
parseSlotNo_ :: String -> Either String C.SlotNo
425+
parseSlotNo_ str = maybe (leftError "Can't read SlotNo" str) (Right . C.SlotNo) $ Read.readMaybe str
426+
427+
-- eitherParseHashBlockHeader :: String -> Either RawBytesHexError (C.Hash C.BlockHeader) -- cardano-api-1.35.4:Cardano.Api.SerialiseRaw.
428+
eitherParseHashBlockHeader = C.deserialiseFromRawBytesHex (C.proxyToAsType Proxy) . C8.pack
429+
430+
eitherParseHashBlockHeader_ :: String -> Either String (C.Hash C.BlockHeader)
431+
eitherParseHashBlockHeader_ = either (Left . show) Right . eitherParseHashBlockHeader
432+
433+
leftError :: String -> String -> Either String a
434+
leftError label str = Left $ label <> ": '" <> str <> "'"
309435

310436
-- * Sqlite
311437

‎mafoc/src/Mafoc/Indexers/EpochNonce.hs

+88
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
{-# LANGUAGE LambdaCase #-}
2+
{-# LANGUAGE NamedFieldPuns #-}
3+
{-# LANGUAGE OverloadedStrings #-}
4+
module Mafoc.Indexers.EpochNonce where
5+
6+
import Cardano.Api qualified as C
7+
import Cardano.Ledger.Shelley.API qualified as Ledger
8+
import Data.String (fromString)
9+
import Database.SQLite.Simple qualified as SQL
10+
import Options.Applicative qualified as Opt
11+
12+
import Mafoc.CLI qualified as Opt
13+
import Mafoc.Core (DbPathAndTableName, Indexer (Event, Runtime, State, checkpoint, initialize, persistMany, toEvent),
14+
LocalChainsyncConfig, NodeConfig, initializeLedgerState, storeLedgerState)
15+
import Marconi.ChainIndex.Indexers.EpochState qualified as Marconi
16+
17+
data EpochNonce = EpochNonce
18+
{ chainsyncConfig :: LocalChainsyncConfig NodeConfig
19+
, dbPathAndTableName :: DbPathAndTableName
20+
} deriving Show
21+
22+
data EpochNonceEvent = EpochNonceEvent
23+
{ epochNo :: C.EpochNo
24+
, epochNonce :: Ledger.Nonce
25+
}
26+
27+
parseCli :: Opt.ParserInfo EpochNonce
28+
parseCli = Opt.info (Opt.helper <*> cli) $ Opt.fullDesc
29+
<> Opt.progDesc "epochnonce"
30+
<> Opt.header "epochnonce - Index epoch nonces"
31+
where
32+
cli :: Opt.Parser EpochNonce
33+
cli = EpochNonce
34+
<$> Opt.mkCommonLocalChainsyncConfig Opt.commonNodeConnectionAndConfig
35+
<*> Opt.commonDbPathAndTableName
36+
37+
instance Indexer EpochNonce where
38+
39+
type Event EpochNonce = EpochNonceEvent
40+
41+
data Runtime EpochNonce = Runtime
42+
{ sqlConnection :: SQL.Connection
43+
, tableName :: String
44+
, ledgerCfg :: Marconi.ExtLedgerCfg_
45+
}
46+
data State EpochNonce = State
47+
{ extLedgerState :: Marconi.ExtLedgerState_
48+
, maybePreviousEpochNo :: Maybe C.EpochNo
49+
}
50+
51+
toEvent (Runtime{ledgerCfg}) state blockInMode = return (State newExtLedgerState maybeEpochNo, maybeEvent)
52+
where
53+
newExtLedgerState = Marconi.applyBlock ledgerCfg (extLedgerState state) blockInMode
54+
maybeEpochNo = Marconi.getEpochNo newExtLedgerState
55+
epochNonce = Marconi.getEpochNonce newExtLedgerState
56+
maybeEvent :: Maybe EpochNonceEvent
57+
maybeEvent = case maybeEpochNo of
58+
Just epochNo -> case maybePreviousEpochNo state of
59+
Just previousEpochNo -> case epochNo - previousEpochNo of
60+
1 -> Just $ EpochNonceEvent epochNo epochNonce
61+
0 -> Nothing
62+
invalidEpochDiff -> error $ "EpochNonce indexer: assumption violated: epoch changed by " <> show invalidEpochDiff <> " instead of expected 0 or 1."
63+
Nothing -> Just $ EpochNonceEvent epochNo epochNonce
64+
Nothing -> Nothing
65+
66+
initialize EpochNonce{chainsyncConfig, dbPathAndTableName} trace = do
67+
(extLedgerState, epochNo, chainsyncRuntime', sqlCon, tableName, ledgerConfig) <-
68+
initializeLedgerState chainsyncConfig trace dbPathAndTableName sqliteInit "epoch_nonce"
69+
return ( State extLedgerState epochNo
70+
, chainsyncRuntime'
71+
, Runtime sqlCon tableName ledgerConfig)
72+
73+
persistMany Runtime{sqlConnection, tableName} events = sqliteInsert sqlConnection tableName events
74+
75+
checkpoint Runtime{ledgerCfg} State{extLedgerState} slotNoBhh = storeLedgerState ledgerCfg slotNoBhh extLedgerState
76+
77+
-- * Sqlite
78+
79+
sqliteInit :: SQL.Connection -> String -> IO ()
80+
sqliteInit c tableName = SQL.execute_ c $
81+
" CREATE TABLE IF NOT EXISTS " <> fromString tableName <> " \
82+
\ ( epoch_no INT NOT NULL \
83+
\ , nonce BLOB NOT NULL ) "
84+
85+
sqliteInsert :: SQL.Connection -> String -> [EpochNonceEvent] -> IO ()
86+
sqliteInsert c tableName events = SQL.executeMany c
87+
("INSERT INTO " <> fromString tableName <>" (epoch_no, nonce) VALUES (?, ?)")
88+
(map (\(EpochNonceEvent epochNo nonce) -> (epochNo, nonce)) events)

0 commit comments

Comments
 (0)