add envDbIsolationState for when syncing and following

Author: Cmdv

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

@@ -144,7 +144,7 @@ runDbSync metricsSetters iomgr trce params syncNodeConfigFromFile abortOnPanic =
   whenJust (enpMaybeRollback params) $ \slotNo ->
     void $ unsafeRollback trce (txOutConfigToTableType txOutConfig) pgConfig slotNo
 
-  -- This runMigration is ONLY for delayed migrations during sync (like indexes)
+  -- These migrations will be ran when near the tip of the chain eg: indexes.
   let runNearTipMigration mode = do
         msg <- DB.getMaintenancePsqlConf pgConfig
         logInfo trce $ "Running NearTip database migrations in mode " <> textShow mode

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

@@ -339,6 +339,7 @@ mkSyncEnv metricSetters trce dbEnv syncOptions protoInfo nw nwMagic systemStart
   oarq <- newTBQueueIO 1000
   epochVar <- newTVarIO initCurrentEpochNo
   epochStatistics <- initEpochStatistics
+  dbIsolationStateVar <- newTVarIO DB.SyncLagging  -- For database transaction isolation optimisation
   ledgerEnvType <-
     case (enpMaybeLedgerStateDir syncNP, hasLedger' syncNodeConfigFromFile) of
       (Just dir, True) ->
@@ -370,6 +371,7 @@ mkSyncEnv metricSetters trce dbEnv syncOptions protoInfo nw nwMagic systemStart
       , envDbConstraints = dbCNamesVar
       , envCurrentEpochNo = epochVar
       , envIndexes = indexesVar
+      , envDbIsolationState = dbIsolationStateVar
       , envLedgerEnv = ledgerEnvType
       , envNetworkMagic = nwMagic
       , envOffChainPoolResultQueue = oprq

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

@@ -50,6 +50,7 @@ data SyncEnv = SyncEnv
   , envCurrentEpochNo :: !(StrictTVar IO CurrentEpochNo)
   , envIndexes :: !(StrictTVar IO Bool)
   , envBootstrap :: !(StrictTVar IO Bool)
+  , envDbIsolationState :: !(StrictTVar IO DB.SyncState)
   , envLedgerEnv :: !LedgerEnv
   , envNetworkMagic :: !NetworkMagic
   , envOffChainPoolResultQueue :: !(StrictTBQueue IO OffChainPoolResult)

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

@@ -1,4 +1,3 @@
-{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 
@@ -63,16 +62,18 @@ data ThreadChannels = ThreadChannels
 --
 -- This is the primary transaction runner for sequential database operations in db-sync.
 -- All operations within the ExceptT stack are executed atomically in one database transaction.
+-- Accepts an optional isolation level (Nothing uses RepeatableRead default).
 runDbSyncTransaction ::
   forall m a.
   (MonadUnliftIO m, HasCallStack) =>
   Trace IO Text ->
   DB.DbEnv ->
+  Maybe DB.IsolationLevel ->
   ExceptT SyncNodeError DB.DbM a ->
   m (Either SyncNodeError a)
-runDbSyncTransaction tracer dbEnv exceptTAction = do
+runDbSyncTransaction tracer dbEnv mIsolationLevel exceptTAction = do
   -- Catch database exceptions and convert to Either
-  eResult <- liftIO $ try $ DB.runDbTransLogged tracer dbEnv (runExceptT exceptTAction)
+  eResult <- liftIO $ try $ DB.runDbTransLogged tracer dbEnv mIsolationLevel (runExceptT exceptTAction)
   case eResult of
     Left (dbErr :: DB.DbSessionError) -> do
       pure $ Left $ SNErrDbSessionErr mkSyncNodeCallStack dbErr
@@ -88,7 +89,7 @@ runDbSyncTransactionNoLogging ::
   m (Either SyncNodeError a)
 runDbSyncTransactionNoLogging dbEnv exceptTAction = do
   let dbAction = runExceptT exceptTAction
-  eResult <- liftIO $ try $ DB.runDbTransSilent dbEnv dbAction
+  eResult <- liftIO $ try $ DB.runDbTransSilent dbEnv Nothing dbAction
   case eResult of
     Left (dbErr :: DB.DbSessionError) -> do
       pure $ Left $ SNErrDbSessionErr mkSyncNodeCallStack dbErr
@@ -135,7 +136,7 @@ runDbSyncTransactionPool ::
   m (Either SyncNodeError a)
 runDbSyncTransactionPool tracer dbEnv exceptTAction = do
   let dbAction = runExceptT exceptTAction
-  eResult <- liftIO $ try $ DB.runDbPoolTransLogged tracer dbEnv dbAction -- Use pool
+  eResult <- liftIO $ try $ DB.runDbPoolTransLogged tracer dbEnv Nothing dbAction -- Use pool
   case eResult of
     Left (dbErr :: DB.DbSessionError) -> do
       pure $ Left $ SNErrDbSessionErr mkSyncNodeCallStack dbErr

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

@@ -46,15 +46,17 @@ import Cardano.DbSync.Rollback
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util
 import Cardano.DbSync.Util.Constraint (addConstraintsIfNotExist)
+import Control.Concurrent.Class.MonadSTM.Strict (readTVarIO)
 
 insertListBlocks ::
   SyncEnv ->
   [CardanoBlock] ->
   IO (Either SyncNodeError ())
 insertListBlocks syncEnv blocks = do
+  isolationLevel <- determineIsolationLevel syncEnv
   -- stop at the exact block number if the option is set
   case sioStopAtBlock $ dncInsertOptions $ envSyncNodeConfig syncEnv of
-    Nothing -> runDbSyncTransaction (getTrace syncEnv) (envDbEnv syncEnv) $ do
+    Nothing -> runDbSyncTransaction (getTrace syncEnv) (envDbEnv syncEnv) isolationLevel $ do
       traverse_ (applyAndInsertBlockMaybe syncEnv (getTrace syncEnv)) blocks
     Just targetBlock ->
       insertListBlocksWithStopCondition syncEnv blocks targetBlock
@@ -70,7 +72,8 @@ insertListBlocksWithStopCondition syncEnv blocks targetBlock = do
   -- Check if we hit the stop condition in this batch
   let hitStopCondition = any (\cblk -> unBlockNo (blockNo cblk) >= targetBlock) blocks
   -- Process the blocks in transaction
-  result <- runDbSyncTransaction (getTrace syncEnv) (envDbEnv syncEnv) $ do
+  isolationLevel <- determineIsolationLevel syncEnv
+  result <- runDbSyncTransaction (getTrace syncEnv) (envDbEnv syncEnv) isolationLevel $ do
     traverse_ (applyAndInsertBlockMaybe syncEnv (getTrace syncEnv)) blocksToProcess
   -- If we hit the stop condition and transaction succeeded, shutdown
   case result of
@@ -243,6 +246,14 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
 
     blkNo = headerFieldBlockNo $ getHeaderFields cblk
 
+-- | Determine isolation level based on current sync state
+determineIsolationLevel :: SyncEnv -> IO (Maybe DB.IsolationLevel)
+determineIsolationLevel syncEnv = do
+  syncState <- readTVarIO (envDbIsolationState syncEnv)
+  pure $ case syncState of
+    DB.SyncLagging -> Just DB.ReadCommitted    -- Syncing: use ReadCommitted for performance
+    DB.SyncFollowing -> Nothing                -- Following: use default RepeatableRead for consistency
+
 isWithinTwoMin :: SlotDetails -> Bool
 isWithinTwoMin sd = isSyncedWithinSeconds sd 120 == SyncFollowing
 

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

@@ -45,7 +45,8 @@ insertValidateByronGenesisDist ::
   Byron.Config ->
   ExceptT SyncNodeError IO ()
 insertValidateByronGenesisDist syncEnv (NetworkName networkName) cfg = do
-  ExceptT $ runDbSyncTransaction (getTrace syncEnv) (envDbEnv syncEnv) insertAction
+  -- Genesis insertion is always syncing, use ReadCommitted for better performance
+  ExceptT $ runDbSyncTransaction (getTrace syncEnv) (envDbEnv syncEnv) (Just DB.ReadCommitted) insertAction
   where
     tracer = getTrace syncEnv
 

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

@@ -14,11 +14,13 @@ module Cardano.DbSync.Era.Universal.Insert.Grouped (
   insertBlockGroupedData,
   insertReverseIndex,
   resolveTxInputs,
+  resolveTxInputsBulk,
   resolveScriptHash,
   mkmaTxOuts,
 ) where
 
 import qualified Data.List as List
+import qualified Data.Map.Strict as Map
 import qualified Data.Text as Text
 
 import Cardano.BM.Trace (logWarning)
@@ -240,6 +242,7 @@ resolveTxInputs syncEnv hasConsumed needsValue groupedOutputs txIn = do
       DB.VCTxOutW cTxOut -> (txIn, VC.txOutCoreTxId cTxOut, Left txIn, Nothing)
       DB.VATxOutW vTxOut _ -> (txIn, VA.txOutAddressTxId vTxOut, Left txIn, Nothing)
 
+
 resolveRemainingInputs ::
   [ExtendedTxIn] ->
   [(DB.TxOutIdW, ExtendedTxOut)] ->

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

@@ -31,7 +31,7 @@ import Cardano.DbSync.Types
 
 import Cardano.DbSync.Error (SyncNodeError)
 import Cardano.DbSync.Metrics (setDbEpochSyncDuration, setDbEpochSyncNumber)
-import Control.Concurrent.Class.MonadSTM.Strict (readTVarIO)
+import Control.Concurrent.Class.MonadSTM.Strict (readTVarIO, writeTVar)
 import Control.Monad.Extra (whenJust)
 import qualified Data.Map.Strict as Map
 import qualified Data.Set as Set
@@ -76,8 +76,11 @@ insertNewEpochLedgerEvents syncEnv currentEpochNo@(EpochNo curEpoch) =
           currentTime <- liftIO getCurrentTime
           -- Get current epoch statistics
           epochStats <- liftIO $ readTVarIO (envEpochStatistics syncEnv)
+          -- Update the database isolation state for transaction optimisation
+          let syncState = toSyncState ss
+          liftIO $ atomically $ writeTVar (envDbIsolationState syncEnv) syncState
           -- Insert the epoch sync time into the database
-          insertEpochSyncTime en (toSyncState ss) epochStats currentTime
+          insertEpochSyncTime en syncState epochStats currentTime
           -- Text of the epoch sync time
           let epochDurationText = formatEpochDuration (elsStartTime epochStats) currentTime
 

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

@@ -53,22 +53,25 @@ import Cardano.Db.Types (DbEnv (..), DbM (..))
 -- This is the primary runner used for cardano-db-sync block processing.
 -- Wraps all operations in a single database transaction with full ACID guarantees.
 -- Automatically handles BEGIN/COMMIT/ROLLBACK and provides comprehensive logging.
+-- Accepts an optional isolation level (defaults to RepeatableRead).
 runDbTransLogged ::
   MonadUnliftIO m =>
   Trace IO Text ->
   DbEnv ->
+  Maybe IsolationLevel ->  -- Optional isolation level
   DbM a ->
   m a
-runDbTransLogged tracer dbEnv action = do
+runDbTransLogged tracer dbEnv mIsolationLevel action = do
   result <- liftIO $ HsqlS.run transactionSession (dbConnection dbEnv)
   case result of
     Left sessionErr -> do
       liftIO $ logWarning tracer $ "Database transaction error: " <> Text.pack (show sessionErr)
       throwIO $ DbSessionError mkDbCallStack ("Database transaction error: " <> formatSessionError sessionErr)
     Right dbResult -> pure dbResult
   where
+    isolationLevel = fromMaybe RepeatableRead mIsolationLevel
     transactionSession = do
-      HsqlS.statement () (beginTransactionStmt RepeatableRead)
+      HsqlS.statement () (beginTransactionStmt isolationLevel)
 
       result <- liftIO $ try @SomeException $ runIohkLogging tracer $ liftIO $ runReaderT (runDbM action) dbEnv
       case result of
@@ -83,21 +86,24 @@ runDbTransLogged tracer dbEnv action = do
 --
 -- Same transaction guarantees as runDbTransLogged but without logging.
 -- Useful for performance-critical operations or testing where log output isn't needed.
+-- Accepts an optional isolation level (defaults to RepeatableRead).
 runDbTransSilent ::
   MonadUnliftIO m =>
   DbEnv ->
+  Maybe IsolationLevel ->  -- Optional isolation level
   DbM a ->
   m a
-runDbTransSilent dbEnv action = do
+runDbTransSilent dbEnv mIsolationLevel action = do
   runNoLoggingT $ do
     result <- liftIO $ HsqlS.run transactionSession (dbConnection dbEnv)
     case result of
       Left sessionErr ->
         throwIO $ DbSessionError mkDbCallStack ("Database transaction error: " <> formatSessionError sessionErr)
       Right dbResult -> pure dbResult
   where
+    isolationLevel = fromMaybe RepeatableRead mIsolationLevel
     transactionSession = do
-      HsqlS.statement () (beginTransactionStmt RepeatableRead)
+      HsqlS.statement () (beginTransactionStmt isolationLevel)
 
       result <- liftIO $ try @SomeException $ runReaderT (runDbM action) dbEnv
       case result of
@@ -159,13 +165,15 @@ runDbDirectSilent dbEnv action = do
 -- Uses a connection from the pool rather than the main DbEnv connection.
 -- Wraps operations in a transaction with logging. Designed for concurrent operations
 -- where multiple threads need independent database connections.
+-- Accepts an optional isolation level (defaults to RepeatableRead).
 runDbPoolTransLogged ::
   MonadUnliftIO m =>
   Trace IO Text ->
   DbEnv ->
+  Maybe IsolationLevel ->  -- Optional isolation level
   DbM a ->
   m a
-runDbPoolTransLogged tracer dbEnv action = do
+runDbPoolTransLogged tracer dbEnv mIsolationLevel action = do
   case dbPoolConnection dbEnv of
     Nothing -> throwIO $ DbSessionError mkDbCallStack "No connection pool available in DbEnv"
     Just pool -> do
@@ -176,8 +184,9 @@ runDbPoolTransLogged tracer dbEnv action = do
             Left sessionErr -> throwIO $ DbSessionError mkDbCallStack ("Pool transaction error: " <> formatSessionError sessionErr)
             Right dbResult -> pure dbResult
   where
+    isolationLevel = fromMaybe RepeatableRead mIsolationLevel
     transactionSession conn = do
-      HsqlS.statement () (beginTransactionStmt RepeatableRead)
+      HsqlS.statement () (beginTransactionStmt isolationLevel)
       result <- liftIO $ try @SomeException $ do
         let tempDbEnv = createDbEnv conn (dbPoolConnection dbEnv) (dbTracer dbEnv)
         runReaderT (runDbM action) tempDbEnv
@@ -233,7 +242,7 @@ runDbStandaloneTransSilent source action = do
     HsqlCon.release
     ( \connection -> do
         let dbEnv = createDbEnv connection Nothing Nothing
-        runDbTransSilent dbEnv action
+        runDbTransSilent dbEnv Nothing action
     )
 
 -- | Standalone runner without transaction management

cardano-db/src/Cardano/Db/Statement/Variants/TxOut.hs

@@ -15,6 +15,7 @@ import qualified Data.Text as Text
 import qualified Data.Text.Encoding as TextEnc
 import qualified Hasql.Decoders as HsqlD
 import qualified Hasql.Encoders as HsqlE
+import qualified Hasql.Pipeline as HsqlP
 import qualified Hasql.Session as HsqlSes
 import qualified Hasql.Statement as HsqlStmt
 
@@ -31,7 +32,6 @@ import Cardano.Db.Statement.Function.InsertBulk (insertBulk)
 import Cardano.Db.Statement.Function.Query (adaDecoder, countAll)
 import Cardano.Db.Statement.Types (DbInfo (..), Entity (entityVal))
 import Cardano.Db.Types (Ada (..), DbLovelace, DbM, DbWord64, dbLovelaceDecoder)
-import qualified Hasql.Pipeline as HsqlP
 
 --------------------------------------------------------------------------------
 -- TxOut