start adding pool queries

Author: Cmdv

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

@@ -204,10 +204,12 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runIndexesMigrationFnc syncN
     ( \dbConn -> do
         runOrThrowIO $ runExceptT $ do
           let isLogingEnabled = dncEnableDbLogging syncNodeConfigFromFile
-              dbEnv =
+          -- Create connection pool for parallel operations
+          pool <- liftIO $ DB.createHasqlConnectionPool [dbConnSetting] 4 -- 4 connections for reasonable parallelism
+          let dbEnv =
                 if isLogingEnabled
-                  then DB.DbEnv dbConn isLogingEnabled (Just trce)
-                  else DB.DbEnv dbConn isLogingEnabled Nothing
+                  then DB.createDbEnv dbConn pool (Just trce)
+                  else DB.createDbEnv dbConn pool Nothing
           genCfg <- readCardanoGenesisConfig syncNodeConfigFromFile
           isJsonbInSchema <- liftDbError $ DB.queryJsonbInSchemaExists dbConn
           logProtocolMagicId trce $ genesisProtocolMagicId genCfg

cardano-db-sync/src/Cardano/DbSync/Era/Byron/Genesis.hs

@@ -47,9 +47,9 @@ insertValidateByronGenesisDist ::
 insertValidateByronGenesisDist syncEnv (NetworkName networkName) cfg = do
   -- Setting this to True will log all 'Persistent' operations which is great
   -- for debugging, but otherwise *way* too chatty.
-  if DB.dbEnableLogging $ envDbEnv syncEnv
-    then liftDbIO $ DB.runDbIohkLogging tracer (envDbEnv syncEnv) insertAction
-    else liftDbIO $ DB.runDbIohkNoLogging (envDbEnv syncEnv) insertAction
+  case DB.dbTracer $ envDbEnv syncEnv of
+    Just trce -> liftDbIO $ DB.runDbIohkLogging trce (envDbEnv syncEnv) insertAction
+    Nothing -> liftDbIO $ DB.runDbIohkNoLogging (envDbEnv syncEnv) insertAction
   where
     tracer = getTrace syncEnv
 

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Genesis.hs

@@ -69,9 +69,9 @@ insertValidateShelleyGenesisDist syncEnv networkName cfg shelleyInitiation = do
     liftIO $ logError tracer $ show SNErrIgnoreShelleyInitiation
     throwError SNErrIgnoreShelleyInitiation
 
-  if DB.dbEnableLogging $ envDbEnv syncEnv
-    then liftDbIO $ DB.runDbIohkLogging tracer (envDbEnv syncEnv) (insertAction prunes)
-    else liftDbIO $ DB.runDbIohkNoLogging (envDbEnv syncEnv) (insertAction prunes)
+  case DB.dbTracer $ envDbEnv syncEnv of
+    Just trce -> liftDbIO $ DB.runDbIohkLogging trce (envDbEnv syncEnv) (insertAction prunes)
+    Nothing -> liftDbIO $ DB.runDbIohkNoLogging (envDbEnv syncEnv) (insertAction prunes)
   where
     tracer = getTrace syncEnv
 

cardano-db-sync/src/Cardano/DbSync/Era/Universal/Insert/Grouped.hs

@@ -162,7 +162,7 @@ insertBlockGroupedData syncEnv grouped = do
     a2 <- async $ pure $ prepareMetadataProcessing syncEnv grouped
     a3 <- async $ pure $ prepareMintProcessing syncEnv grouped
     a4 <- async $ pure $ chunksOf maxBulkSize $ etoTxOut . fst <$> groupedTxOut grouped
-    
+
     r1 <- wait a1
     r2 <- wait a2
     r3 <- wait a3
@@ -172,20 +172,20 @@ insertBlockGroupedData syncEnv grouped = do
   -- Sequential TxOut processing (generates required IDs)
   txOutIds <- concat <$> mapM (DB.insertBulkTxOut disInOut) txOutChunks
 
-  -- PHASE 3: Execute independent operations (TxIn, Metadata, Mint) in parallel
+  -- Execute independent operations (TxIn, Metadata, Mint) in parallel
   txInIds <- executePreparedTxIn preparedTxIn
-  
-  -- PHASE 4: Pipeline TxOut-dependent operations (MaTxOut + UTxO consumption)
+
+  -- TxOut-dependent operations (MaTxOut + UTxO consumption)
   maTxOutIds <- processMaTxOuts syncEnv txOutIds grouped
-  
-  -- PHASE 5: Execute remaining independent operations in parallel
+
+  -- Execute remaining independent operations in parallel with pools
   liftIO $ do
-    a1 <- async $ DB.runDbActionIO (envDbEnv syncEnv) (executePreparedMetadata preparedMetadata)
-    a2 <- async $ DB.runDbActionIO (envDbEnv syncEnv) (executePreparedMint preparedMint)
+    a1 <- async $ DB.runPoolDbAction (envDbEnv syncEnv) (executePreparedMetadata preparedMetadata)
+    a2 <- async $ DB.runPoolDbAction (envDbEnv syncEnv) (executePreparedMint preparedMint)
     _ <- wait a1
     void $ wait a2
-  
-  -- PHASE 6: Process UTxO consumption (depends on txOutIds)
+
+  -- Process UTxO consumption (depends on txOutIds)
   processUtxoConsumption syncEnv grouped txOutIds
 
   pure $ makeMinId syncEnv txInIds txOutIds maTxOutIds
@@ -363,7 +363,7 @@ data PreparedTxIn = PreparedTxIn
   , ptiSkip :: !Bool
   }
 
--- | Prepared Metadata data for async execution  
+-- | Prepared Metadata data for async execution
 data PreparedMetadata = PreparedMetadata
   { pmChunks :: ![[DB.TxMetadata]]
   , pmRemoveJsonb :: !Bool
@@ -399,7 +399,7 @@ prepareMintProcessing _syncEnv grouped =
 
 -- | Execute prepared TxIn operations
 executePreparedTxIn :: MonadIO m => PreparedTxIn -> DB.DbAction m [DB.TxInId]
-executePreparedTxIn prepared = 
+executePreparedTxIn prepared =
   if ptiSkip prepared
     then pure []
     else concat <$> mapM DB.insertBulkTxIn (ptiChunks prepared)
@@ -418,8 +418,9 @@ executePreparedMint prepared =
 processMaTxOuts :: MonadIO m => SyncEnv -> [DB.TxOutIdW] -> BlockGroupedData -> DB.DbAction m [DB.MaTxOutIdW]
 processMaTxOuts syncEnv txOutIds grouped = do
   let txOutVariantType = getTxOutVariantType syncEnv
-      maTxOuts = concatMap (mkmaTxOuts txOutVariantType) $
-                   zip txOutIds (snd <$> groupedTxOut grouped)
+      maTxOuts =
+        concatMap (mkmaTxOuts txOutVariantType) $
+          zip txOutIds (snd <$> groupedTxOut grouped)
       maTxOutChunks = chunksOf maxBulkSize maTxOuts
   concat <$> mapM DB.insertBulkMaTxOut maTxOutChunks
 
@@ -428,7 +429,7 @@ processUtxoConsumption :: MonadIO m => SyncEnv -> BlockGroupedData -> [DB.TxOutI
 processUtxoConsumption syncEnv grouped txOutIds = do
   let tracer = getTrace syncEnv
       txOutVariantType = getTxOutVariantType syncEnv
-      
+
   whenConsumeOrPruneTxOut syncEnv $ do
     -- Resolve remaining inputs
     etis <- resolveRemainingInputs (groupedTxIn grouped) $ zip txOutIds (fst <$> groupedTxOut grouped)

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

@@ -240,7 +240,11 @@ runFetchOffChainPoolThread syncEnv syncNodeConfigFromFile = do
       HsqlC.release
       ( \dbConn -> forever $ do
           -- Create a new DbEnv for this thread
-          let dbEnv = DB.DbEnv dbConn (dncEnableDbLogging syncNodeConfigFromFile) $ Just trce
+          pool <- DB.createHasqlConnectionPool [connSetting] 4 -- 4 connections for reasonable parallelism
+          let dbEnv =
+                if dncEnableDbLogging syncNodeConfigFromFile
+                  then DB.createDbEnv dbConn pool (Just trce)
+                  else DB.createDbEnv dbConn pool Nothing
               -- Create a new SyncEnv with the new DbEnv but preserving all other fields
               threadSyncEnv = syncEnv {envDbEnv = dbEnv}
           tDelay
@@ -275,7 +279,11 @@ runFetchOffChainVoteThread syncEnv syncNodeConfigFromFile = do
       HsqlC.release
       ( \dbConn -> do
           -- Create a new DbEnv for this thread
-          let dbEnv = DB.DbEnv dbConn (dncEnableDbLogging syncNodeConfigFromFile) $ Just trce
+          pool <- DB.createHasqlConnectionPool [connSetting] 4 -- 4 connections for reasonable parallelism
+          let dbEnv =
+                if dncEnableDbLogging syncNodeConfigFromFile
+                  then DB.createDbEnv dbConn pool (Just trce)
+                  else DB.createDbEnv dbConn pool Nothing
           -- Create a new SyncEnv with the new DbEnv but preserving all other fields
           let threadSyncEnv = syncEnv {envDbEnv = dbEnv}
           -- Use the thread-specific SyncEnv for all operations

cardano-db/src/Cardano/Db/Run.hs

@@ -2,6 +2,7 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RankNTypes #-}
 
 module Cardano.Db.Run where
 
@@ -127,7 +128,8 @@ runDbActionWithIsolation dbEnv isolationLevel action = do
         Right _ -> do
           -- Run action with async exception protection via onException
           -- If interrupted (Ctrl+C), the onException handler will rollback
-          result <- onException
+          result <-
+            onException
               (restore (runInIO $ runReaderT (runExceptT (runDbAction action)) dbEnv))
               (restore $ rollbackTransaction dbEnv)
           case result of
@@ -196,16 +198,9 @@ runPoolDbIohkLogging ::
   m (Either DbError a)
 runPoolDbIohkLogging connPool tracer action = do
   conn <- liftIO $ withResource connPool pure
-  let dbEnv = mkDbEnv conn
+  let dbEnv = createDbEnv conn connPool (Just tracer)
   runIohkLogging tracer $
     runDbActionWithIsolation dbEnv RepeatableRead action
-  where
-    mkDbEnv conn =
-      DbEnv
-        { dbConnection = conn
-        , dbEnableLogging = True
-        , dbTracer = Just tracer
-        }
 
 runDbNoLogging :: MonadUnliftIO m => PGPassSource -> DbAction m a -> m a
 runDbNoLogging source action = do
@@ -217,9 +212,11 @@ runDbNoLogging source action = do
     bracket
       (acquireConnection [connSetting])
       HsqlCon.release
-      ( \connection -> runInIO $ do
-          let dbEnv = DbEnv connection False Nothing
-          runDbConnWithIsolation action dbEnv RepeatableRead
+      ( \connection -> do
+          pool <- createHasqlConnectionPool [connSetting] 4 -- 4 connections for reasonable parallelism
+          runInIO $ do
+            let dbEnv = createDbEnv connection pool Nothing
+            runDbConnWithIsolation action dbEnv RepeatableRead
       )
 
 runDbNoLoggingEnv :: MonadUnliftIO m => DbAction m a -> m a
@@ -235,7 +232,8 @@ runWithConnectionNoLogging source action = do
     (acquireConnection [connSetting])
     HsqlCon.release
     ( \connection -> do
-        let dbEnv = DbEnv connection False Nothing
+        pool <- createHasqlConnectionPool [connSetting] 4 -- 4 connections for reasonable parallelism
+        let dbEnv = createDbEnv connection pool Nothing
         runNoLoggingT $ runDbConnWithIsolation action dbEnv RepeatableRead
     )
 
@@ -295,3 +293,33 @@ createHasqlConnectionPool settings numConnections = do
         Left err -> throwIO $ userError $ "Connection error: " <> show err
         Right conn -> pure conn
     releaseConn = HsqlCon.release
+
+-- Helper to create DbEnv with both single connection and pool
+createDbEnv :: HsqlCon.Connection -> Pool HsqlCon.Connection -> Maybe (Trace IO Text) -> DbEnv
+createDbEnv conn pool tracer =
+  DbEnv
+    { dbConnection = conn
+    , dbPoolConnection = pool
+    , dbTracer = tracer
+    }
+
+-- Pool-aware database action runners for async operations
+runPoolDbAction :: forall a m. MonadUnliftIO m => DbEnv -> DbAction m a -> m a
+runPoolDbAction dbEnv action = do
+  withRunInIO $ \runInIO -> do
+    conn <- withResource (dbPoolConnection dbEnv) pure
+    let poolDbEnv = dbEnv {dbConnection = conn, dbTracer = Nothing} -- No logging for pool operations to avoid contention
+    result <- runInIO $ runReaderT (runExceptT (runDbAction action)) poolDbEnv
+    case result of
+      Left err -> throwIO err
+      Right val -> pure val
+
+runPoolDbActionWithLogging :: forall a m. MonadUnliftIO m => DbEnv -> DbAction m a -> m a
+runPoolDbActionWithLogging dbEnv action = do
+  withRunInIO $ \runInIO -> do
+    conn <- withResource (dbPoolConnection dbEnv) pure
+    let poolDbEnv = dbEnv {dbConnection = conn} -- Keep original logging settings
+    result <- runInIO $ runReaderT (runExceptT (runDbAction action)) poolDbEnv
+    case result of
+      Left err -> throwIO err
+      Right val -> pure val

cardano-db/src/Cardano/Db/Statement/Function/Core.hs

@@ -16,7 +16,7 @@ where
 import Cardano.BM.Trace (logInfo)
 import Cardano.Db.Error (DbCallStack (..), DbError (..))
 import Cardano.Db.Types (DbAction (..), DbEnv (..))
-import Cardano.Prelude (MonadError (..), MonadIO (..), Text, ask, for_, when)
+import Cardano.Prelude (MonadError (..), MonadIO (..), Text, ask)
 import qualified Data.Text as Text
 import Data.Time (diffUTCTime, getCurrentTime)
 import GHC.Stack (HasCallStack, SrcLoc (..), callStack, getCallStack)
@@ -57,11 +57,7 @@ import qualified Hasql.Session as HsqlS
 runDbSession :: MonadIO m => DbCallStack -> HsqlS.Session a -> DbAction m a
 runDbSession dbCallStack@DbCallStack {..} session = DbAction $ do
   dbEnv <- ask
-  let logMsg msg =
-        when (dbEnableLogging dbEnv) $
-          for_ (dbTracer dbEnv) $
-            \tracer -> liftIO $ logInfo tracer msg
-      locationInfo =
+  let locationInfo =
         " Function: "
           <> dbCsFncName
           <> " at "
@@ -71,15 +67,15 @@ runDbSession dbCallStack@DbCallStack {..} session = DbAction $ do
           <> ":"
           <> Text.pack (show dbCsLine)
 
-  if dbEnableLogging dbEnv
-    then do
+  case dbTracer dbEnv of
+    Nothing -> run dbEnv
+    Just tracer -> do
       start <- liftIO getCurrentTime
       result <- run dbEnv
       end <- liftIO getCurrentTime
       let duration = diffUTCTime end start
-      logMsg $ "Query: " <> dbCsFncName <> locationInfo <> " in " <> Text.pack (show duration)
+      liftIO $ logInfo tracer $ "Query: " <> dbCsFncName <> locationInfo <> " in " <> Text.pack (show duration)
       pure result
-    else run dbEnv
   where
     run dbEnv = do
       result <- liftIO $ HsqlS.run session (dbConnection dbEnv)

cardano-db/src/Cardano/Db/Types.hs

@@ -30,6 +30,7 @@ import Data.Either (fromRight)
 import Data.Fixed (Micro, showFixed)
 import Data.Functor.Contravariant ((>$<))
 import Data.Int (Int64)
+import Data.Pool (Pool)
 import Data.Scientific (Scientific (..), scientific, toBoundedInteger)
 import Data.Text (Text)
 import qualified Data.Text as Text
@@ -60,7 +61,7 @@ newtype DbAction m a = DbAction
 ----------------------------------------------------------------------------
 data DbEnv = DbEnv
   { dbConnection :: !HsqlCon.Connection
-  , dbEnableLogging :: !Bool
+  , dbPoolConnection :: !(Pool HsqlCon.Connection)
   , dbTracer :: !(Maybe (Trace IO Text))
   }