Merge pull request #2076 from IntersectMBO/2041-fix-isolation-commits

2041 - fix isolation commits

Author: kderme

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
 
@@ -245,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 DB.ReadCommitted (DB.querySlotHash slotNo)
 
 hasLedgerState :: SyncEnv -> Bool
 hasLedgerState syncEnv =
@@ -256,7 +258,7 @@ hasLedgerState syncEnv =
 getDbLatestBlockInfo :: DB.DbEnv -> IO (Maybe TipInfo)
 getDbLatestBlockInfo dbEnv = do
   runMaybeT $ do
-    block <- MaybeT $ DB.runDbDirectSilent dbEnv DB.queryLatestBlock
+    block <- MaybeT $ DB.runDbTransSilent dbEnv DB.ReadCommitted DB.queryLatestBlock
     -- The EpochNo, SlotNo and BlockNo can only be zero for the Byron
     -- era, but we need to make the types match, hence `fromMaybe`.
     pure $
@@ -267,6 +269,14 @@ getDbLatestBlockInfo dbEnv = do
         , bBlockNo = BlockNo . fromMaybe 0 $ DB.blockBlockNo block
         }
 
+-- | 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
   mblk <- getDbLatestBlockInfo (envDbEnv env)
@@ -314,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 DB.ReadCommitted DB.queryRewardAndEpochStakeConstraints
   cache <-
     if soptCache syncOptions
       then
@@ -506,7 +516,7 @@ getBootstrapInProgress ::
   DB.DbEnv ->
   IO Bool
 getBootstrapInProgress trce bootstrapFlag dbEnv = do
-  DB.runDbDirectSilent dbEnv $ do
+  DB.runDbTransSilent dbEnv DB.ReadCommitted $ do
     ems <- DB.queryAllExtraMigrations
     let btsState = DB.bootstrapState ems
     case (bootstrapFlag, btsState) of

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

@@ -89,7 +89,7 @@ runDbSyncTransactionNoLogging ::
   m (Either SyncNodeError a)
 runDbSyncTransactionNoLogging dbEnv exceptTAction = do
   let dbAction = runExceptT exceptTAction
-  eResult <- liftIO $ try $ DB.runDbTransSilent dbEnv Nothing dbAction
+  eResult <- liftIO $ try $ DB.runDbTransSilent dbEnv DB.RepeatableRead dbAction
   case eResult of
     Left (dbErr :: DB.DbSessionError) -> do
       pure $ Left $ SNErrDbSessionErr mkSyncNodeCallStack dbErr

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,7 +23,7 @@ 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 Cardano.DbSync.OffChain.Http
@@ -248,9 +248,10 @@ runFetchOffChainPoolThread syncEnv = do
               threadSyncEnv = syncEnv {envDbEnv = dbEnv}
           forever $ do
             tDelay
-            -- load the offChain vote work queue using the db
+            -- Use dynamic isolation level based on sync state
+            mIsolationLevel <- determineIsolationLevel syncEnv
             _ <-
-              DB.runDbDirectLogged trce dbEnv $
+              DB.runDbTransLogged trce dbEnv mIsolationLevel $
                 loadOffChainPoolWorkQueue trce (envOffChainPoolWorkQueue threadSyncEnv)
             poolq <- atomically $ flushTBQueue (envOffChainPoolWorkQueue threadSyncEnv)
             manager <- Http.newManager tlsManagerSettings
@@ -284,9 +285,10 @@ runFetchOffChainVoteThread syncEnv = do
           -- Use the thread-specific SyncEnv for all operations
           forever $ do
             tDelay
-            -- load the offChain vote work queue using the db
+            -- Use dynamic isolation level based on sync state
+            mIsolationLevel <- determineIsolationLevel syncEnv
             _ <-
-              DB.runDbDirectLogged trce dbEnv $
+              DB.runDbTransLogged trce dbEnv mIsolationLevel $
                 loadOffChainVoteWorkQueue trce (envOffChainVoteWorkQueue threadSyncEnv)
             voteq <- atomically $ flushTBQueue (envOffChainVoteWorkQueue threadSyncEnv)
             now <- liftIO Time.getPOSIXTime

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) DB.ReadCommitted $ runExceptT action
   where
     trce = getTrace syncEnv
 

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

@@ -90,21 +90,19 @@ runDbTransLogged tracer dbEnv mIsolationLevel action = do
 runDbTransSilent ::
   MonadUnliftIO m =>
   DbEnv ->
-  Maybe IsolationLevel -> -- Optional isolation level
+  IsolationLevel ->
   DbM a ->
   m a
-runDbTransSilent dbEnv mIsolationLevel action = do
+runDbTransSilent dbEnv isolationLevel 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 isolationLevel)
-
       result <- liftIO $ try @SomeException $ runReaderT (runDbM action) dbEnv
       case result of
         Left err -> do
@@ -272,7 +270,7 @@ runDbStandaloneTransSilent source action = do
     HsqlCon.release
     ( \connection -> do
         let dbEnv = createDbEnv connection Nothing Nothing
-        runDbTransSilent dbEnv Nothing action
+        runDbTransSilent dbEnv RepeatableRead action
     )
 
 -- | Standalone runner without transaction management