diff --git a/marconi-chain-index/marconi-chain-index.cabal b/marconi-chain-index/marconi-chain-index.cabal index 6fd0486dbc..fa2928520e 100644 --- a/marconi-chain-index/marconi-chain-index.cabal +++ b/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 diff --git a/marconi-chain-index/test-compare-cardano-db-sync/DBUtils.hs b/marconi-chain-index/test-compare-cardano-db-sync/DBUtils.hs new file mode 100644 index 0000000000..fbaf388696 --- /dev/null +++ b/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!" diff --git a/marconi-chain-index/test-compare-cardano-db-sync/EpochState.hs b/marconi-chain-index/test-compare-cardano-db-sync/EpochState.hs new file mode 100644 index 0000000000..fabaa0a660 --- /dev/null +++ b/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 diff --git a/marconi-chain-index/test-compare-cardano-db-sync/Spec.hs b/marconi-chain-index/test-compare-cardano-db-sync/Spec.hs index 3ac10ca9ce..196cf7af40 100644 --- a/marconi-chain-index/test-compare-cardano-db-sync/Spec.hs +++ b/marconi-chain-index/test-compare-cardano-db-sync/Spec.hs @@ -1,8 +1,3 @@ -{-# 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 @@ -14,9 +9,9 @@ - CARDANO_NODE_SOCKET_PATH - CARDANO_NODE_CONFIG_PATH + - DBSYNC_PG_URL - MARCONI_DB_DIRECTORY_PATH - - DBSYNC_PGPASSWORD: The default password for cardano-db-sync's postgres database is in its repo in the file: config/secrets/postgres_password - - NETWORK_MAGIC: "mainnet" or number + - NETWORK_MAGIC And then run the command: @@ -29,41 +24,10 @@ -} module Main 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.Environment (lookupEnv) -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 EpochState qualified +import Utxo qualified -import Hedgehog ((===)) -import Hedgehog qualified as H import Test.Tasty (TestTree, defaultMain, testGroup) -import Test.Tasty.Hedgehog (testPropertyNamed) main :: IO () main = defaultMain tests @@ -72,184 +36,6 @@ 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 + [ EpochState.tests + , Utxo.tests ] - -{- | 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 - -{- | Connect to cardano-db-sync postgres with password from - DBSYNC_PGPASSWORD env variable. --} -getDbSyncPgConnection :: H.PropertyT IO PG.Connection -getDbSyncPgConnection = do - pgPassword <- envOrFail "DBSYNC_PGPASSWORD" - liftIO $ - PG.connect $ - PG.ConnectInfo - { PG.connectHost = "localhost" - , PG.connectPort = 5432 - , PG.connectUser = "postgres" - , PG.connectPassword = pgPassword - , PG.connectDatabase = "cexplorer" - } - --- | Get string from the environment or fail test with instruction. -envOrFail :: String -> H.PropertyT IO String -envOrFail str = - liftIO (lookupEnv str) >>= \case - Just v -> return v - Nothing -> fail $ str <> " environment variable not set!" - -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 diff --git a/marconi-chain-index/test-compare-cardano-db-sync/Utxo.hs b/marconi-chain-index/test-compare-cardano-db-sync/Utxo.hs new file mode 100644 index 0000000000..2bfce79cec --- /dev/null +++ b/marconi-chain-index/test-compare-cardano-db-sync/Utxo.hs @@ -0,0 +1,429 @@ +{- +-- | The purpose of this module is to compare the Utxos between cardano-db-sync and Marconi +Assumptions: + +1. Both marconi and cardano-db-sync are synched with cardano node +2. the follwoing environemnt variales are set: + - 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 + + To run this specific test: +@ +cabal test marconi-chain-index-test-compare-cardano-db-sync --flags="-ci" --test-option=--pattern="Utxo" +-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} +{-# OPTIONS_GHC -Wno-orphans #-} + +module Utxo where + +import Control.Concurrent.Async (Concurrently (Concurrently), runConcurrently) +import Control.Concurrent.STM.TMChan (TMChan, closeTMChan, newTMChanIO, readTMChan, writeTMChan) +import Control.Monad (void) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Coerce (coerce) +import Data.Proxy (Proxy (Proxy)) +import Data.Text (Text) +import Data.Word (Word64) +import Database.PostgreSQL.Simple qualified as PG +import Database.PostgreSQL.Simple.FromField qualified as PG +import Database.PostgreSQL.Simple.FromRow qualified as PG +import Database.PostgreSQL.Simple.ToField qualified as PG +import Database.PostgreSQL.Simple.ToRow qualified as PG +import Database.SQLite.Simple qualified as SQLite +import Database.SQLite.Simple.ToField qualified as SQLite +import GHC.Conc (atomically, forkIO) +import Streaming (Of, Stream) +import Streaming.Prelude qualified as S +import Text.RawString.QQ (r) + +import Cardano.Api qualified as C +import DBUtils qualified as UtxoDb +import Hedgehog (Property, property, withTests) +import Hedgehog qualified +import Marconi.ChainIndex.Orphans () +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.Hedgehog (testPropertyNamed) + +tests :: TestTree +tests = + testGroup + "Utxo" + [ testPropertyNamed + "Compare all Utxo addresses between Marconi SQLite database and cardano-db-sync" + "marconiUtxoAddressesAreSubsetOfDbSyncUtxosAddresses" + marconiUtxoAddressesAreSubsetOfDbSyncUtxosAddresses + ] + +findDbSyncAddressInMarconi :: Property +findDbSyncAddressInMarconi = + withTests 1 $ + property $ do + (UtxoDb.DbConnection sqliteConn _) <- UtxoDb.mkDbConnection + liftIO $ printAllAddresses sqliteConn + Hedgehog.success + +{- +-- The purpose of this test is to verify addresses in marconi SQLite unspent_transaction are in cardano-db-sync +-- Note: This test does not address marconi in-memeory utxos. +-- Relaxing this requirement has the following pro/conss: +-- Pros: + * cardano-node does not have to be up + * the test is repeatable and simpler to verify as it avoids potential race conditions of running marconi and cardano-db-sync in parallel +-- Cons: + * the test is not comprehensive +-- Upon failure, we print a list of delta-addresses, addresses that are in dbSync and not in Marconi +-- +-- Assumptions: +- marconi and cardano-db-sync are synchronised with cardano node +-} +marconiUtxoAddressesAreSubsetOfDbSyncUtxosAddresses :: Property +marconiUtxoAddressesAreSubsetOfDbSyncUtxosAddresses = + withTests 1 $ + property $ do + dbs@(UtxoDb.DbConnection sqliteConn _) <- UtxoDb.mkDbConnection + liftIO $ initDBs dbs + + [maxSlot] :: [Integer] <- liftIO $ SQLite.query_ sqliteConn "SELECT MAX(slotNo) FROM marconiUtxos" + liftIO $ mkDeltaAddressesTable sqliteConn maxSlot + + [SQLite.Only deltaMarconiDbSyncCount] :: [SQLite.Only Int] <- + liftIO $ SQLite.query_ sqliteConn "SELECT COUNT(1) FROM deltaAddresses" + [SQLite.Only dbSyncUtxoCount] :: [SQLite.Only Int] <- + liftIO $ SQLite.query_ sqliteConn "SELECT COUNT(1) FROM dbSyncUtxos" + [SQLite.Only marconiUtxoCount] :: [SQLite.Only Int] <- + liftIO $ SQLite.query_ sqliteConn "SELECT COUNT(1) FROM marconiUtxos" + + liftIO . putStrLn $ + "Address Utxo Marconi and cardano-db-sync comparason where maximum slotNo = " + <> show maxSlot + <> "\nnumber Utxos in marconi database: " + <> show marconiUtxoCount + <> "\nnumber Utxos in cardano-db-sync database: " + <> show dbSyncUtxoCount + <> "\ndelta addresses between marconi and cardano-db-sync: " + <> show deltaMarconiDbSyncCount + + Hedgehog.annotate $ + "deltaMarconiDbSyncCount: " + <> show deltaMarconiDbSyncCount + <> "\nmarconiUtxoCount: " + <> show marconiUtxoCount + <> "\ndbsyncUtxoCount: " + Hedgehog.assert $ marconiUtxoCount > 1 + Hedgehog.assert $ dbSyncUtxoCount > 1 + if deltaMarconiDbSyncCount /= 0 + then (liftIO $ printAllAddresses sqliteConn) >> Hedgehog.failure -- proof that for the slotNo range, there are equal number of Address-Utxos between db-sync and marconi + else Hedgehog.success + where + mkDeltaAddressesTable :: SQLite.Connection -> Integer -> IO () + mkDeltaAddressesTable sqliteConn maxSlot = do + void $ SQLite.execute_ sqliteConn "DROP TABLE IF EXISTS deltaAddresses" + void $ + SQLite.execute + sqliteConn + [r|CREATE TABLE deltaAddresses AS + SELECT marconi.address, marconi.slotNo + FROM marconiUtxos marconi + LEFT JOIN dbSyncUtxos dbsync + ON (marconi.address = dbsync.address) + WHERE dbsync.address IS NULL AND dbsync.slotNo <= ?|] + (SQLite.Only maxSlot) + +{- +initialize a SQL workflow: + * create a postgres view of utxos that includes slot_no + * stream the above to marconi SQLight + * create subsets of the above db-sync originated utxos, so that we may compare marconi utxos with those of db-sync + * create a delta table that contains utxo addresses in marconi that do not exist in db-sync + * This test requires both marconi and db-sync have been synchronised with cardano-node +-} +initDBs :: UtxoDb.DbConnection -> IO () +initDBs (UtxoDb.DbConnection sqliteConn pgConn) = + let mkSQLiteMarconiUtxos :: IO () + mkSQLiteMarconiUtxos = + liftIO $ + SQLite.execute_ sqliteConn "DROP TABLE IF EXISTS marconiUtxos" + >> SQLite.execute_ + sqliteConn + [r|CREATE TABLE IF NOT EXISTS marconiUtxos AS + SELECT + u.address + , u.txId + , u.txIx + , u.datum + , u.datumHash + , u.value + , u.inlineScript + , u.inlineScriptHash + , u.slotNo + , u.blockHash + FROM unspent_transactions u + LEFT JOIN spent s ON + u.txId = s.txId AND u.txIx = s.txIx + WHERE + s.txId IS NULL AND s.txIx IS NULL |] + + mkPGDbSyncUtxosView :: IO () -- create a dbsync view with slotNo to the utxo_view + mkPGDbSyncUtxosView = + void $ + PG.execute_ + pgConn + [r|CREATE OR REPLACE VIEW utxos_v AS + SELECT + tx.hash txId, + v.index txOutIndex, + v.address_raw, + v.value, + block.slot_no slotNo, + block.hash blockHash, + block.block_no blockNo, + v.data_hash datumHash + FROM utxo_view as v + INNER JOIN tx on v.tx_id = tx.id + INNER JOIN block on block.id = tx.block_id|] + + mkSQLiteDbSyncUtxos :: IO () + mkSQLiteDbSyncUtxos = + liftIO $ + SQLite.execute_ sqliteConn "DROP TABLE IF EXISTS dbSyncUtxos" + >> SQLite.execute_ + sqliteConn + [r|CREATE TABLE IF NOT EXISTS dbSyncUtxos + ( txId BLOB NOT NULL + , txIndex INT + , address BLOB NOT NULL + , slotNo INT + , blockHash BLOB + , blockNo INT + , datumHash BLOB)|] + + loadUtxosFromDbSync :: IO () + loadUtxosFromDbSync = do + liftIO $ + S.mapM_ + ( SQLite.execute + sqliteConn + [r|INSERT INTO dbSyncUtxos + ( txId + , txIndex + , address + , slotNo + , blockHash + , blockNo + , datumHash) VALUES (?, ?, ?, ?, ?, ?, ?)|] + ) + ( sourceQuery_ + pgConn + [r|SELECT + v.txId + , v.txOutIndex + , v.address_raw + , v.slotNo + , v.blockHash + , v.blockNo + , v.datumHash + FROM utxos_v v|] + :: S.Stream (S.Of DbSyncUtxo) IO () + ) + in do + void $ + runConcurrently $ + (,,) + <$> Concurrently mkPGDbSyncUtxosView + <*> Concurrently mkSQLiteMarconiUtxos + <*> Concurrently mkSQLiteDbSyncUtxos + loadUtxosFromDbSync + +-- * SQL mappings +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.ToRow Integer where + toRow = PG.toRow + +instance PG.FromRow Integer where + fromRow = PG.field + +instance SQLite.ToRow Word64 where + toRow :: Word64 -> [SQLite.SQLData] + toRow = SQLite.toRow + +newtype Address = Address {unAddress :: C.AddressAny} + +instance SQLite.FromRow Address where + fromRow = Address <$> SQLite.field + +instance SQLite.FromRow Word64 where + fromRow = SQLite.fromRow + +instance SQLite.FromRow Text where + fromRow = SQLite.fromRow + +instance PG.ToRow Word64 where + toRow = PG.toRow + +instance PG.FromRow Word64 where + fromRow = PG.fromRow + +instance PG.FromField (C.Hash C.BlockHeader) where + fromField f meta = + let cantDeserialise = PG.returnError PG.ConversionFailed f "Cannot deserialise address." + in ( PG.fromField f meta + >>= ( \case + Right a -> pure a + Left _ -> cantDeserialise + ) + . C.deserialiseFromRawBytes (C.proxyToAsType Proxy) + ) + +instance PG.FromField C.AddressAny where + fromField f meta = + let cantDeserialise = PG.returnError PG.ConversionFailed f "Cannot deserialise address." + in ( PG.fromField f meta + >>= ( \case + Right a -> pure a + Left _ -> cantDeserialise + ) + . C.deserialiseFromRawBytes C.AsAddressAny + ) + +instance PG.FromField C.TxId where + fromField f meta = + PG.fromField f meta + >>= ( \case + Right a -> return a + Left err -> PG.returnError PG.ConversionFailed f $ "Can't parse C.TxId , error: " <> show err + ) + . C.deserialiseFromRawBytes (C.proxyToAsType Proxy) + +instance PG.ToField (C.Hash C.BlockHeader) where + toField = PG.toField . C.serialiseToRawBytes + +instance PG.ToField C.AddressAny where + toField = PG.toField . C.serialiseToRawBytes + +instance PG.FromField (C.Hash C.ScriptData) where + fromField f meta = + PG.fromField f meta + >>= ( \case + Right a -> return a + Left err -> PG.returnError PG.ConversionFailed f $ "Can't parse C.AsHash C.AsScriptData , error: " <> show err + ) + . C.deserialiseFromRawBytes (C.AsHash C.AsScriptData) + +instance SQLite.FromRow DbSyncUtxo where + fromRow = + DbSyncUtxo + <$> SQLite.field + <*> SQLite.field + <*> SQLite.field + <*> SQLite.field + <*> SQLite.field + <*> SQLite.field + <*> SQLite.field + +instance PG.FromField Word64 where + fromField f meta = fromIntegral @Integer <$> PG.fromField f meta + +instance PG.FromField C.SlotNo where + fromField f meta = fromIntegral @Integer <$> PG.fromField f meta + +instance PG.ToField C.SlotNo where + toField = PG.toField . coerce @C.SlotNo @Word64 + +instance SQLite.ToRow DbSyncUtxo where + toRow u = + [ SQLite.toField (dbTxId u) + , SQLite.toField (dbTxIndex u) + , SQLite.toField (dbAddress u) + , SQLite.toField (dbSlotNo u) + , SQLite.toField (dbBlockHash u) + , SQLite.toField (dbBlockNo u) + , SQLite.toField (dbDatumHash u) + ] + +instance PG.FromRow DbSyncUtxo where + fromRow = + DbSyncUtxo + <$> PG.field + <*> PG.field + <*> PG.field + <*> PG.field + <*> PG.field + <*> PG.field + <*> PG.field + +deriving newtype instance PG.ToRow C.SlotNo + +-- * create a streaming source/destination to stream a supperset of the utxo_view from cardano-db-sync to marconi-sqlite + +-- | create a streaming source +mkStreamingSource + :: (MonadIO m) + => ((r -> IO ()) -> IO ()) + -> Stream (Of r) m () +mkStreamingSource action = + let chanSource + :: MonadIO m + => TMChan r + -> Stream (Of r) m () + chanSource chan = loop + where + loop = do + p <- liftIO $ atomically $ readTMChan chan + case p of + Just x -> S.yield x >> loop + Nothing -> pure () + in do + chan <- liftIO newTMChanIO + _ <- liftIO $ + forkIO $ do + action $ atomically . writeTMChan chan + void (liftIO $ atomically $ closeTMChan chan) + chanSource chan + +-- | Stream rows from cardano-db-sync, no param substition +sourceQuery_ + :: (MonadIO m, PG.FromRow r) + => PG.Connection + -> PG.Query + -> Stream (Of r) m () +sourceQuery_ conn q = mkStreamingSource $ PG.forEach_ conn q + +-- | Stream rows from cardano-db-sync, with param substition +sourceQuery + :: (PG.ToRow params, PG.FromRow r, MonadIO m) + => PG.Connection + -> PG.Query + -> params + -> Stream (Of r) m () +sourceQuery conn q params = mkStreamingSource $ PG.forEach conn q params + +data DbSyncUtxo = DbSyncUtxo + { dbTxId :: C.TxId + , dbTxIndex :: Word64 + , dbAddress :: C.AddressAny + , dbSlotNo :: C.SlotNo + , dbBlockHash :: C.Hash C.BlockHeader + , dbBlockNo :: Word64 + , dbDatumHash :: !(Maybe (C.Hash C.ScriptData)) + } + deriving (Eq, Show) + +-- | print the list of offending addresses. These are addresses in db-sync and not in marconi that have utxos +printAllAddresses :: SQLite.Connection -> IO () +printAllAddresses c = do + putStrLn "starting the test" + addresses <- + SQLite.query_ c "SELECT address from deltaAddresses" :: IO [Address] + let addrs :: [Text] = C.serialiseAddress . unAddress <$> addresses + putStrLn "/nAddresses that are in Marconi Utxo and not in dbSync utxo_view\n" + print addrs