reduce redundant transactions

Author: Cmdv

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

@@ -1,3 +1,4 @@
+{-# LANGUAGE ApplicativeDo #-}
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
@@ -43,6 +44,7 @@ module Cardano.DbSync.Api (
   generateNewEpochEvents,
   logDbState,
   convertToPoint,
+  determineIsolationLevel,
 )
 where
 
@@ -91,8 +93,6 @@ import Cardano.DbSync.Ledger.Types (HasLedgerEnv (..), LedgerStateFile (..), Sna
 import Cardano.DbSync.LocalStateQuery
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util
-import qualified Hasql.Pipeline as HsqlP
-import qualified Hasql.Session as HsqlSes
 
 setConsistentLevel :: SyncEnv -> ConsistentLevel -> IO ()
 setConsistentLevel env cst = do
@@ -247,7 +247,7 @@ getInsertOptions :: SyncEnv -> InsertOptions
 getInsertOptions = soptInsertOptions . envOptions
 
 getSlotHash :: DB.DbEnv -> SlotNo -> IO [(SlotNo, ByteString)]
-getSlotHash backend = DB.runDbDirectSilent backend . DB.querySlotHash
+getSlotHash backend slotNo = DB.runDbTransSilent backend (Just DB.ReadCommitted) (DB.querySlotHash slotNo)
 
 hasLedgerState :: SyncEnv -> Bool
 hasLedgerState syncEnv =
@@ -269,18 +269,13 @@ getDbLatestBlockInfo dbEnv = do
         , bBlockNo = BlockNo . fromMaybe 0 $ DB.blockBlockNo block
         }
 
--- | Pipeline block and points queries in a single transaction for better performance
-loadBlockAndPointsData :: DB.DbEnv -> IO (Maybe DB.Block, [(Maybe Word64, ByteString)])
-loadBlockAndPointsData dbEnv =
-  DB.runDbTransSilent dbEnv (Just DB.ReadCommitted) $ do
-    (mBlockEntity, points) <- DB.runSession DB.mkDbCallStack $
-      HsqlSes.pipeline $ do
-        blk <- HsqlP.statement () DB.queryLatestBlockStmt
-        pts <- HsqlP.statement () DB.queryLatestPointsStmt
-        pure (blk, pts)
-    
-    let mBlock = DB.entityVal <$> mBlockEntity
-    pure (mBlock, points)
+-- | Determine isolation level based on sync state
+determineIsolationLevel :: SyncEnv -> IO (Maybe DB.IsolationLevel)
+determineIsolationLevel syncEnv = do
+  syncState <- readTVarIO (envDbIsolationState syncEnv)
+  pure $ case syncState of
+    DB.SyncLagging -> Just DB.ReadCommitted
+    DB.SyncFollowing -> Nothing
 
 getDbTipBlockNo :: SyncEnv -> IO (Point.WithOrigin BlockNo)
 getDbTipBlockNo env = do
@@ -329,7 +324,7 @@ mkSyncEnv ::
   Bool ->
   IO SyncEnv
 mkSyncEnv metricSetters trce dbEnv syncOptions protoInfo nw maxLovelaceSupply nwMagic systemStart syncNodeConfigFromFile syncNP runNearTipMigrationFnc isJsonbInSchema = do
-  dbCNamesVar <- newTVarIO =<< DB.runDbDirectSilent dbEnv DB.queryRewardAndEpochStakeConstraints
+  dbCNamesVar <- newTVarIO =<< DB.runDbTransSilent dbEnv (Just DB.ReadCommitted) DB.queryRewardAndEpochStakeConstraints
   cache <-
     if soptCache syncOptions
       then
@@ -466,7 +461,7 @@ getLatestPoints env = do
       verifySnapshotPoint env snapshotPoints
     NoLedger _ -> do
       -- Brings the 5 latest.
-      lastPoints <- DB.runDbDirectSilent (envDbEnv env) DB.queryLatestPoints
+      lastPoints <- DB.runDbTransSilent (envDbEnv env) (Just DB.ReadCommitted) DB.queryLatestPoints
       pure $ mapMaybe convert lastPoints
   where
     convert (Nothing, _) = Nothing
@@ -521,7 +516,7 @@ getBootstrapInProgress ::
   DB.DbEnv ->
   IO Bool
 getBootstrapInProgress trce bootstrapFlag dbEnv = do
-  DB.runDbDirectSilent dbEnv $ do
+  DB.runDbTransSilent dbEnv (Just DB.ReadCommitted) $ do
     ems <- DB.queryAllExtraMigrations
     let btsState = DB.bootstrapState ems
     case (bootstrapFlag, btsState) of

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

@@ -46,7 +46,6 @@ 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 ->
@@ -249,13 +248,6 @@ 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/OffChain.hs

@@ -23,10 +23,9 @@ module Cardano.DbSync.OffChain (
 
 import Cardano.BM.Trace (Trace, logInfo)
 import qualified Cardano.Db as DB
-import Cardano.DbSync.Api
+import Cardano.DbSync.Api (determineIsolationLevel, getInsertOptions, getTrace)
 import Cardano.DbSync.Api.Types (InsertOptions (..), SyncEnv (..))
 import Cardano.DbSync.Config.Types
-import qualified Cardano.DbSync.Default as Default
 import Cardano.DbSync.OffChain.Http
 import Cardano.DbSync.OffChain.Query
 import qualified Cardano.DbSync.OffChain.Vote.Types as Vote
@@ -249,9 +248,8 @@ runFetchOffChainPoolThread syncEnv = do
               threadSyncEnv = syncEnv {envDbEnv = dbEnv}
           forever $ do
             tDelay
-            -- Determine isolation level based on sync state (dynamic optimization)
-            mIsolationLevel <- Default.determineIsolationLevel syncEnv
-            -- Use transaction runner with dynamic isolation level to eliminate redundant BEGIN/COMMIT
+            -- Use dynamic isolation level based on sync state
+            mIsolationLevel <- determineIsolationLevel syncEnv
             _ <-
               DB.runDbTransLogged trce dbEnv mIsolationLevel $
                 loadOffChainPoolWorkQueue trce (envOffChainPoolWorkQueue threadSyncEnv)
@@ -287,9 +285,8 @@ runFetchOffChainVoteThread syncEnv = do
           -- Use the thread-specific SyncEnv for all operations
           forever $ do
             tDelay
-            -- Determine isolation level based on sync state (dynamic optimization)
-            mIsolationLevel <- Default.determineIsolationLevel syncEnv
-            -- Use transaction runner with dynamic isolation level to eliminate redundant BEGIN/COMMIT
+            -- Use dynamic isolation level based on sync state
+            mIsolationLevel <- determineIsolationLevel syncEnv
             _ <-
               DB.runDbTransLogged trce dbEnv mIsolationLevel $
                 loadOffChainVoteWorkQueue trce (envOffChainVoteWorkQueue threadSyncEnv)

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

@@ -68,7 +68,7 @@ rollbackFromBlockNo syncEnv blkNo = do
 
 prepareRollback :: SyncEnv -> CardanoPoint -> Tip CardanoBlock -> IO (Either SyncNodeError Bool)
 prepareRollback syncEnv point serverTip = do
-  DB.runDbDirectSilent (envDbEnv syncEnv) $ runExceptT action
+  DB.runDbTransSilent (envDbEnv syncEnv) (Just DB.ReadCommitted) $ runExceptT action
   where
     trce = getTrace syncEnv