Compare Marconi Utxo with cardano-db-sync

marconi-chain-index/marconi-chain-index.cabal

@@ -379,6 +379,10 @@ test-suite marconi-chain-index-test-compare-cardano-db-sync
   type:           exitcode-stdio-1.0
   main-is:        Spec.hs
   hs-source-dirs: test-compare-cardano-db-sync
+  other-modules:
+    DBUtils
+    EpochState
+    Utxo
 
   if flag(ci)
     buildable: False
@@ -419,7 +423,7 @@ test-suite marconi-chain-index-test-compare-cardano-db-sync
   build-depends:
     , aeson
     , async
-    , base                  >=4.9 && <5
+    , base                   >=4.9 && <5
     , base16-bytestring
     , bytestring
     , cborg
@@ -431,10 +435,13 @@ test-suite marconi-chain-index-test-compare-cardano-db-sync
     , mtl
     , optparse-applicative
     , postgresql-simple
+    , postgresql-simple-url
     , prettyprinter
+    , raw-strings-qq
     , serialise
     , sqlite-simple
     , stm
+    , stm-chans
     , streaming
     , tasty
     , tasty-golden

marconi-chain-index/test-compare-cardano-db-sync/DBUtils.hs

@@ -0,0 +1,68 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE TemplateHaskell #-}
+
+module DBUtils where
+
+import Control.Concurrent.STM.TMChan ()
+import Control.Exception (handle)
+import Control.Lens.TH (makeLenses)
+import Control.Monad.IO.Class (liftIO)
+import Database.PostgreSQL.Simple ()
+import Database.PostgreSQL.Simple qualified as PG
+import Database.PostgreSQL.Simple.FromRow ()
+import Database.PostgreSQL.Simple.ToRow ()
+import Database.PostgreSQL.Simple.URL (parseDatabaseUrl)
+import Database.SQLite.Simple qualified as SQLite
+import GHC.Generics (Generic)
+import System.Environment (lookupEnv)
+import System.FilePath (combine)
+
+import Hedgehog qualified
+
+data DbConnection = DbConnection
+  { _dbcSQLite :: SQLite.Connection
+  , _dbcPG :: PG.Connection
+  }
+  deriving (Generic)
+
+$(makeLenses ''DbConnection)
+
+{- | Connect to cardano-db-sync postgres with password from
+ DBSYNC_PG_URL, Postgres Connection URI, env variable, see section 34.1.1.2.Connection URIs, https://www.postgresql.org/docs/current/libpq-connect.html
+-}
+getDbSyncPgConnection :: Hedgehog.PropertyT IO PG.Connection
+getDbSyncPgConnection = do
+  url <- envOrFail "DBSYNC_PG_URL"
+  liftIO $
+    maybe
+      (fail "Failed parsing Postgres Connection URL")
+      PG.connect
+      (parseDatabaseUrl url)
+
+getSQLiteConnection :: Hedgehog.PropertyT IO SQLite.Connection
+getSQLiteConnection = do
+  path <- flip combine "utxo.db" <$> envOrFail "MARCONI_DB_DIRECTORY_PATH"
+  liftIO $
+    handle
+      (\(e :: SQLite.SQLError) -> fail (show e))
+      ( do
+          c <- SQLite.open path
+          SQLite.execute_ c "PRAGMA journal_mode=WAL"
+          pure c
+      )
+
+mkDbConnection :: Hedgehog.PropertyT IO DbConnection
+mkDbConnection = do
+  s <- getSQLiteConnection
+  p <- getDbSyncPgConnection
+  pure $ DbConnection s p
+
+-- | Get string from the environment or fail test with instruction.
+envOrFail :: String -> Hedgehog.PropertyT IO String
+envOrFail str =
+  liftIO $
+    lookupEnv str >>= \case
+      Just v -> return v
+      Nothing -> fail $ str <> " environment variable not set!"

marconi-chain-index/test-compare-cardano-db-sync/EpochState.hs

@@ -0,0 +1,230 @@
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# OPTIONS_GHC -Wno-orphans #-}
+
+{- | Run the Marconi and cardano-db-sync comparison by:
+
+1. Sync up cardano-db-sync
+
+2. Run the EpochState indexer up to sync, possibly using the
+   cardano-node from the cardano-db-sync docker
+
+3. Run this test by setting the env vaiables:
+
+     - CARDANO_NODE_SOCKET_PATH
+     - CARDANO_NODE_CONFIG_PATH
+     - MARCONI_DB_DIRECTORY_PATH
+     - DBSYNC_PG_URL: Postgres URL, see https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING for detail.   cardano-db-sync's postgres database details are in its repo in the file: config/secrets/postgres_password
+     - NETWORK_MAGIC: "mainnet" or number
+
+   And then run the command:
+
+@
+   cabal test marconi-chain-index-test-compare-cardano-db-sync --flag '-ci'
+@
+
+   The --flag '-ci' is there to unset the "ci" cabal flag which is on
+   by default as we don't want to run it on CI.
+-}
+module EpochState where
+
+import Control.Exception (throw)
+import Control.Monad (forM_)
+import Control.Monad.IO.Class (liftIO)
+import Control.Monad.Trans.Except (ExceptT, runExceptT)
+import Data.ByteString qualified as BS
+import Data.Coerce (coerce)
+import Data.Map.Strict qualified as Map
+import Data.Ratio (denominator, numerator)
+import Data.Word (Word64)
+import Database.PostgreSQL.Simple qualified as PG
+import Database.PostgreSQL.Simple.FromField qualified as PG
+import Database.PostgreSQL.Simple.ToField qualified as PG
+import System.FilePath ((</>))
+import Text.Read (readMaybe)
+
+import Cardano.Api qualified as C
+import Cardano.Api.Shelley qualified as C
+import Cardano.Crypto.Hash qualified as Crypto
+import Cardano.Ledger.Shelley.API qualified as Ledger
+import Ouroboros.Consensus.Cardano.Block qualified as O
+import Ouroboros.Consensus.Config qualified as O
+import Ouroboros.Consensus.Node qualified as O
+
+import Marconi.ChainIndex.Error qualified as Marconi
+import Marconi.ChainIndex.Indexers.EpochState qualified as EpochState
+import Marconi.ChainIndex.Node.Client.GenesisConfig qualified as GenesisConfig
+import Marconi.ChainIndex.Types (epochStateDbName)
+import Marconi.ChainIndex.Utils qualified as Utils
+import Marconi.Core.Storable qualified as Storable
+
+import DBUtils (envOrFail, getDbSyncPgConnection)
+
+import Hedgehog ((===))
+import Hedgehog qualified as H
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.Hedgehog (testPropertyNamed)
+
+tests :: TestTree
+tests =
+  testGroup
+    "Marconi to cardano-db-sync comparisons"
+    [ testPropertyNamed
+        "Compare all epoch nonces between Marconi and cardano-db-sync"
+        "propEpochNonce"
+        propEpochNonce
+    , testPropertyNamed
+        "Compare all epoch stakepool sizes between Marconi and cardano-db-sync"
+        "propEpochStakepoolSize"
+        propEpochStakepoolSize
+    ]
+
+{- | Connect to cardano-db-sync's postgres instance, get all (EpochNo,
+ Nonce) tuples, query and compare all of these to the one found in
+ Marconi.
+
+ As the number of epochs is low (406 at the time of writing), then
+ all nonces found in postgres are compared.
+-}
+propEpochNonce :: H.Property
+propEpochNonce = H.withTests 1 $ H.property $ do
+  indexer <- openEpochStateIndexer
+  conn <- getDbSyncPgConnection
+  dbSyncEpochNonces <- liftIO $ PG.query_ conn "select epoch_no, nonce from epoch_param order by epoch_no ASC"
+  forM_ dbSyncEpochNonces $ \(epochNo, dbSyncNonce) -> do
+    res <- liftIO $ queryIndexerEpochNonce epochNo indexer
+    case res of
+      Just indexerNonce -> do
+        H.footnote $ "Comparing epoch " <> show epochNo
+        dbSyncNonce === indexerNonce
+      Nothing ->
+        fail $ "Epoch not found in indexer, is it synchronised? Epoch no: " <> show epochNo
+
+queryIndexerEpochNonce :: C.EpochNo -> Storable.State EpochState.EpochStateHandle -> IO (Maybe Ledger.Nonce)
+queryIndexerEpochNonce epochNo indexer = do
+  let query = EpochState.NonceByEpochNoQuery epochNo
+  res' <- throwIndexerError $ Storable.query indexer query
+  case res' of
+    EpochState.NonceByEpochNoResult res -> return $ EpochState.epochNonceRowNonce <$> res
+    _ -> return Nothing
+
+{- | Connect to cardano-db-sync's postgres instance, get minimum and
+ maximum epoch no from epoch_stake table, then compare random 10
+ epoch stakepool sizes to what we have in the indexer.
+-}
+propEpochStakepoolSize :: H.Property
+propEpochStakepoolSize = H.withTests 1 $ H.property $ do
+  conn <- getDbSyncPgConnection
+  indexer <- openEpochStateIndexer
+  [(minEpochNo :: C.EpochNo, maxEpochNo :: C.EpochNo)] <- liftIO $ PG.query_ conn "SELECT min(epoch_no), max(epoch_no) FROM epoch_stake"
+  let compareEpoch epochNo = do
+        dbSyncResult <- liftIO $ dbSyncStakepoolSizes conn epochNo
+        marconiResult <- liftIO $ indexerStakepoolSizes epochNo indexer
+        H.footnote $
+          "Comparing epoch "
+            <> show epochNo
+            <> ", number of stakepools in epoch "
+            <> show (Map.size dbSyncResult)
+        dbSyncResult === marconiResult
+  H.footnote $
+    "Min and max epoch in cardano-db-sync postgres: "
+      <> show (coerce @_ @Word64 minEpochNo)
+      <> " and "
+      <> show (coerce @_ @Word64 maxEpochNo)
+      <> ")"
+  -- We do '+1' because we are interested in the *active* SDD per epoch, whereas db-sync indexes the
+  -- 'set' stake snapshot per epoch.
+  forM_ [minEpochNo + 1 .. maxEpochNo + 1] compareEpoch
+
+dbSyncStakepoolSizes :: PG.Connection -> C.EpochNo -> IO (Map.Map C.PoolId C.Lovelace)
+dbSyncStakepoolSizes conn epochNo = do
+  dbSyncRows :: [(C.PoolId, Rational)] <-
+    liftIO
+      $ PG.query
+        conn
+        "   SELECT ph.hash_raw AS pool_hash             \
+        \        , sum(amount) AS sum_amount            \
+        \     FROM epoch_stake es                       \
+        \     JOIN pool_hash ph ON es.pool_id = ph.id \
+        \    WHERE epoch_no = ?                         \
+        \ GROUP BY epoch_no, pool_hash                  \
+        \ ORDER BY sum_amount desc                      "
+      -- We do that for the same reason as above. The indexer query returns the *active* SDD for epoch
+      -- 'n', so we need to compare it with the db-sync SDD of epoch 'n - 1'.
+      $ PG.Only (epochNo - 1)
+  return $ Map.fromList $ map (\(a, b) -> (a, rationalToLovelace b)) dbSyncRows
+  where
+    rationalToLovelace :: Rational -> C.Lovelace
+    rationalToLovelace n
+      | 1 <- denominator n = fromIntegral $ numerator n
+      | otherwise = error "getEpochStakepoolSizes: This should never happen, lovelace can't be fractional."
+
+indexerStakepoolSizes :: C.EpochNo -> Storable.State EpochState.EpochStateHandle -> IO (Map.Map C.PoolId C.Lovelace)
+indexerStakepoolSizes epochNo indexer = do
+  let query = EpochState.ActiveSDDByEpochNoQuery epochNo
+  result <- throwIndexerError $ Storable.query indexer query
+  case result of
+    EpochState.ActiveSDDByEpochNoResult rows -> return $ Map.fromList $ map toPair rows
+    _ -> return undefined
+  where
+    toPair row = (EpochState.epochSDDRowPoolId row, EpochState.epochSDDRowLovelace row)
+
+openEpochStateIndexer :: H.PropertyT IO (Storable.State EpochState.EpochStateHandle)
+openEpochStateIndexer = do
+  socketPath <- envOrFail "CARDANO_NODE_SOCKET_PATH"
+  nodeConfigPath <- envOrFail "CARDANO_NODE_CONFIG_PATH"
+  dbDir <- envOrFail "MARCONI_DB_DIRECTORY_PATH"
+  networkMagicStr <- envOrFail "NETWORK_MAGIC"
+  networkMagic <- case networkMagicStr of
+    "mainnet" -> return C.Mainnet
+    _ -> case readMaybe networkMagicStr of
+      Nothing -> fail $ "Can't parse network magic: " <> networkMagicStr
+      Just word32 -> return $ C.Testnet $ C.NetworkMagic word32
+  liftIO $ do
+    securityParam <- throwIndexerError $ Utils.querySecurityParam networkMagic socketPath
+    topLevelConfig <- topLevelConfigFromNodeConfig nodeConfigPath
+    let dbPath = dbDir </> epochStateDbName
+        ledgerStateDirPath = dbDir </> "ledgerStates"
+    throwIndexerError $ EpochState.open topLevelConfig dbPath ledgerStateDirPath securityParam
+
+throwIndexerError :: Monad m => ExceptT Marconi.IndexerError m a -> m a
+throwIndexerError action = either throw return =<< runExceptT action
+
+topLevelConfigFromNodeConfig
+  :: FilePath -> IO (O.TopLevelConfig (O.HardForkBlock (O.CardanoEras O.StandardCrypto)))
+topLevelConfigFromNodeConfig nodeConfigPath = do
+  nodeConfigE <- runExceptT $ GenesisConfig.readNetworkConfig (GenesisConfig.NetworkConfigFile nodeConfigPath)
+  nodeConfig <- either (error . show) pure nodeConfigE
+  genesisConfigE <- runExceptT $ GenesisConfig.readCardanoGenesisConfig nodeConfig
+  genesisConfig <- either (error . show . GenesisConfig.renderGenesisConfigError) pure genesisConfigE
+  return $ O.pInfoConfig (GenesisConfig.mkProtocolInfoCardano genesisConfig)
+
+-- * FromField & ToField instances
+
+instance PG.FromField C.EpochNo where
+  fromField f meta = fromIntegral @Integer <$> PG.fromField f meta
+
+instance PG.ToField C.EpochNo where
+  toField = PG.toField . coerce @C.EpochNo @Word64
+
+instance PG.FromField C.Lovelace where
+  fromField f meta = fromIntegral @Integer <$> PG.fromField f meta
+
+instance PG.FromField Ledger.Nonce where
+  fromField f meta =
+    bsToMaybeNonce <$> PG.fromField f meta >>= \case
+      Just a -> return a
+      _ -> PG.returnError PG.ConversionFailed f "Can't parse Nonce"
+    where
+      bsToMaybeNonce :: BS.ByteString -> Maybe Ledger.Nonce
+      bsToMaybeNonce bs = Ledger.Nonce <$> Crypto.hashFromBytes bs
+
+instance PG.FromField C.PoolId where
+  fromField f meta =
+    C.deserialiseFromRawBytes (C.AsHash C.AsStakePoolKey) <$> PG.fromField f meta >>= \case
+      Right a -> return a
+      Left err -> PG.returnError PG.ConversionFailed f $ "Can't parse PoolId, error: " <> show err
+
+deriving newtype instance Real C.EpochNo
+deriving newtype instance Integral C.EpochNo