Add MA thread and dedicated connections for datum and scripts

Not done yet for Address variant

Author: kderme

cardano-db-sync/cardano-db-sync.cabal

@@ -135,6 +135,7 @@ library
                         Cardano.DbSync.Threads.Database
                         Cardano.DbSync.Threads.EpochStake
                         Cardano.DbSync.Threads.Ledger
+                        Cardano.DbSync.Threads.MultiAsset
                         Cardano.DbSync.Threads.Rewards
                         Cardano.DbSync.Threads.Stake
                         Cardano.DbSync.Threads.TxInResolve
@@ -217,6 +218,7 @@ library
                       , prometheus
                       , psqueues
                       , random-shuffle
+                      , resourcet
                       , resource-pool
                       , scientific
                       , serialise

cardano-db-sync/src/Cardano/DbSync.hs

@@ -29,7 +29,7 @@ import qualified Cardano.Crypto as Crypto
 import qualified Cardano.Db as DB
 import qualified Cardano.Db as Db
 import Cardano.DbSync.Api
-import Cardano.DbSync.Api.Types (InsertOptions (..), RunMigration, SyncEnv (..), SyncOptions (..), envLedgerEnv)
+import Cardano.DbSync.Api.Types
 import Cardano.DbSync.Config (configureLogging)
 import Cardano.DbSync.Config.Cardano
 import Cardano.DbSync.Config.Types
@@ -54,7 +54,7 @@ import Control.Monad.Extra (whenJust)
 import qualified Data.Strict.Maybe as Strict
 import qualified Data.Text as Text
 import Data.Version (showVersion)
-import Database.Persist.Postgresql (ConnectionString, withPostgresqlConn)
+import Database.Persist.Postgresql (ConnectionString)
 import qualified Ouroboros.Consensus.HardFork.Simple as HardFork
 import Ouroboros.Network.NodeToClient (IOManager, withIOManager)
 import Paths_cardano_db_sync (version)
@@ -167,17 +167,17 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
   logInfo trce $ "Using alonzo genesis file from: " <> (show . unGenesisFile $ dncAlonzoGenesisFile syncNodeConfigFromFile)
 
   Db.runIohkLogging trce $
-    withPostgresqlConn dbConnString $
-      \backend -> liftIO $ do
+    withDBSyncConnections dbConnString $
+      \backends -> liftIO $ do
         runOrThrowIO $ runExceptT $ do
           genCfg <- readCardanoGenesisConfig syncNodeConfigFromFile
-          isJsonbInSchema <- queryIsJsonbInSchema backend
+          isJsonbInSchema <- queryIsJsonbInSchema (mainBackend backends)
           logProtocolMagicId trce $ genesisProtocolMagicId genCfg
           syncEnv <-
             ExceptT $
               mkSyncEnvFromConfig
                 trce
-                backend
+                backends
                 dbConnString
                 syncOptions
                 genCfg
@@ -197,7 +197,7 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
           liftIO $ runExtraMigrationsMaybe syncEnv
           unless useLedger $ liftIO $ do
             logInfo trce "Migrating to a no ledger schema"
-            Db.noLedgerMigrations backend trce
+            Db.noLedgerMigrations (mainBackend backends) trce
           insertValidateGenesisDist syncEnv (dncNetworkName syncNodeConfigFromFile) genCfg (useShelleyInit syncNodeConfigFromFile)
 
           -- communication channel between datalayer thread and chainsync-client thread

cardano-db-sync/src/Cardano/DbSync/Api.hs

@@ -26,6 +26,9 @@ module Cardano.DbSync.Api (
   getHasConsumedOrPruneTxOut,
   getSkipTxIn,
   getPrunes,
+  withDBSyncConnections,
+  withScriptConnection,
+  withDatumConnection,
   mkSyncEnvFromConfig,
   getInsertOptions,
   getTrace,
@@ -41,7 +44,7 @@ import qualified Cardano.Chain.Genesis as Byron
 import Cardano.Crypto.ProtocolMagic (ProtocolMagicId (..))
 import qualified Cardano.Db as DB
 import Cardano.DbSync.Api.Types
-import Cardano.DbSync.Cache.Types (CacheCapacity (..), newEmptyCache, newStakeChannels, useNoCache)
+import Cardano.DbSync.Cache.Types (CacheCapacity (..), newEmptyCache, newMAChannels, newStakeChannels, useNoCache)
 import Cardano.DbSync.Config.Cardano
 import Cardano.DbSync.Config.Shelley
 import Cardano.DbSync.Config.Types
@@ -65,9 +68,12 @@ import Control.Concurrent.Class.MonadSTM.Strict (
   writeTVar,
  )
 import qualified Control.Concurrent.Class.MonadSTM.Strict.TBQueue as TBQ
+import Control.Concurrent.MVar
+import Control.Monad.Logger (LoggingT, MonadLoggerIO)
+import Control.Monad.Trans.Resource (MonadUnliftIO)
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (getCurrentTime)
-import Database.Persist.Postgresql (ConnectionString, createPostgresqlPool)
+import Database.Persist.Postgresql (ConnectionString, createPostgresqlPool, withPostgresqlConn)
 import Database.Persist.Sql (SqlBackend)
 import Ouroboros.Consensus.Block.Abstract (BlockProtocol)
 import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
@@ -248,7 +254,7 @@ writePrefetch syncEnv cblock = do
 
 mkSyncEnv ::
   Trace IO Text ->
-  SqlBackend ->
+  DbConnections ->
   ConnectionString ->
   SyncOptions ->
   ProtocolInfo CardanoBlock ->
@@ -259,7 +265,7 @@ mkSyncEnv ::
   SyncNodeParams ->
   RunMigration ->
   IO SyncEnv
-mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runMigrationFnc = do
+mkSyncEnv trce backends connectionString syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runMigrationFnc = do
   dbCNamesVar <- newTVarIO =<< dbConstraintNamesExists backend
   cache <-
     if soptCache syncOptions
@@ -278,8 +284,9 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
   indexesVar <- newTVarIO $ enpForceIndexes syncNP
   bts <- getBootstrapInProgress trce (isTxOutConsumedBootstrap' syncNodeConfigFromFile) backend
   bootstrapVar <- newTVarIO bts
-  -- Offline Pool + Anchor queues
   cChans <- newStakeChannels
+  maChans <- newMAChannels
+  -- Offline Pool + Anchor queues
   opwq <- newTBQueueIO 1000
   oprq <- newTBQueueIO 1000
   oawq <- newTBQueueIO 1000
@@ -309,7 +316,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
 
   pure $
     SyncEnv
-      { envBackend = backend
+      { envBackends = backends
       , envPool = pool
       , envBootstrap = bootstrapVar
       , envCache = cache
@@ -323,6 +330,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
       , envLedgerEnv = ledgerEnvType
       , envNetworkMagic = nwMagic
       , envStakeChans = cChans
+      , envMAChans = maChans
       , envOffChainPoolResultQueue = oprq
       , envOffChainPoolWorkQueue = opwq
       , envOffChainVoteResultQueue = oarq
@@ -335,10 +343,47 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
   where
     hasLedger' = hasLedger . sioLedger . dncInsertOptions
     isTxOutConsumedBootstrap' = isTxOutConsumedBootstrap . sioTxOut . dncInsertOptions
+    backend = mainBackend backends
+
+withDBSyncConnections ::
+  (MonadUnliftIO m, MonadLoggerIO m) =>
+  ConnectionString ->
+  (DbConnections -> m a) ->
+  m a
+withDBSyncConnections connStr action =
+  withPostgresqlConn connStr $ \mainConn ->
+    withPostgresqlConn connStr $ \scriptConn ->
+      withPostgresqlConn connStr $ \datumConn -> do
+        scr <- liftIO $ newMVar scriptConn
+        dt <- liftIO $ newMVar datumConn
+        action $ DbConnections mainConn scr dt
+
+withScriptConnection ::
+  SyncEnv ->
+  ReaderT SqlBackend (LoggingT IO) a ->
+  IO a
+withScriptConnection = withGivenConnection scriptBackend
+
+withDatumConnection ::
+  SyncEnv ->
+  ReaderT SqlBackend (LoggingT IO) a ->
+  IO a
+withDatumConnection = withGivenConnection datumBackend
+
+withGivenConnection ::
+  (DbConnections -> MVar SqlBackend) ->
+  SyncEnv ->
+  ReaderT SqlBackend (LoggingT IO) a ->
+  IO a
+withGivenConnection toConn syncEnv action = do
+  withMVar connVar $ \conn ->
+    DB.runDbLogging conn (getTrace syncEnv) action
+  where
+    connVar = toConn $ envBackends syncEnv
 
 mkSyncEnvFromConfig ::
   Trace IO Text ->
-  SqlBackend ->
+  DbConnections ->
   ConnectionString ->
   SyncOptions ->
   GenesisConfig ->
@@ -347,7 +392,7 @@ mkSyncEnvFromConfig ::
   -- | run migration function
   RunMigration ->
   IO (Either SyncNodeError SyncEnv)
-mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeConfigFromFile syncNodeParams runMigrationFnc =
+mkSyncEnvFromConfig trce backends connectionString syncOptions genCfg syncNodeConfigFromFile syncNodeParams runMigrationFnc =
   case genCfg of
     GenesisCardano _ bCfg sCfg _ _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -374,7 +419,7 @@ mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeCon
           Right
             <$> mkSyncEnv
               trce
-              backend
+              backends
               connectionString
               syncOptions
               (fst $ mkProtocolInfoCardano genCfg [])

cardano-db-sync/src/Cardano/DbSync/Api/Types.hs

@@ -9,13 +9,15 @@ module Cardano.DbSync.Api.Types (
   RunMigration,
   ConsistentLevel (..),
   CurrentEpochNo (..),
+  DbConnections (..),
   Prefetch (..),
   PrefetchTxId (..),
+  envBackend,
   newPrefetch,
 ) where
 
 import qualified Cardano.Db as DB
-import Cardano.DbSync.Cache.Types (CacheStatus, StakeChannels)
+import Cardano.DbSync.Cache.Types (CacheStatus, MAChannels, StakeChannels)
 import Cardano.DbSync.Config.Types (SyncNodeConfig)
 import qualified Cardano.DbSync.Era.Shelley.Generic.Tx.Types as Generic
 import Cardano.DbSync.Ledger.Types (HasLedgerEnv)
@@ -30,6 +32,7 @@ import Cardano.DbSync.Types (
 import Cardano.Slotting.Slot (EpochNo (..))
 import Control.Concurrent.Class.MonadSTM.Strict (StrictTVar, newTBQueueIO, newTVarIO)
 import Control.Concurrent.Class.MonadSTM.Strict.TBQueue (StrictTBQueue)
+import Control.Concurrent.MVar
 import Data.Map (Map)
 import Data.Pool (Pool)
 import qualified Data.Strict.Maybe as Strict
@@ -41,7 +44,7 @@ import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
 
 data SyncEnv = SyncEnv
-  { envBackend :: !SqlBackend
+  { envBackends :: !DbConnections
   , envPool :: !(Pool SqlBackend)
   , envPrefetch :: !Prefetch
   , envCache :: !CacheStatus
@@ -55,6 +58,7 @@ data SyncEnv = SyncEnv
   , envLedgerEnv :: !LedgerEnv
   , envNetworkMagic :: !NetworkMagic
   , envStakeChans :: !StakeChannels
+  , envMAChans :: !MAChannels
   , envOffChainPoolResultQueue :: !(StrictTBQueue IO OffChainPoolResult)
   , envOffChainPoolWorkQueue :: !(StrictTBQueue IO OffChainPoolWorkQueue)
   , envOffChainVoteResultQueue :: !(StrictTBQueue IO OffChainVoteResult)
@@ -65,6 +69,9 @@ data SyncEnv = SyncEnv
   , envSystemStart :: !SystemStart
   }
 
+envBackend :: SyncEnv -> SqlBackend
+envBackend = mainBackend . envBackends
+
 data SyncOptions = SyncOptions
   { soptEpochAndCacheEnabled :: !Bool
   , soptAbortOnInvalid :: !Bool
@@ -108,6 +115,12 @@ newtype CurrentEpochNo = CurrentEpochNo
   { cenEpochNo :: Strict.Maybe EpochNo
   }
 
+data DbConnections = DbConnections
+  { mainBackend :: !SqlBackend
+  , scriptBackend :: !(MVar SqlBackend)
+  , datumBackend :: !(MVar SqlBackend)
+  }
+
 data PrefetchTxId = PrefetchTxIdBlock CardanoBlock | PrefetchTxIdBlocks [CardanoBlock]
 
 data Prefetch = Prefetch