Commit secondary transactions before the main one

Author: kderme

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

@@ -29,6 +29,7 @@ module Cardano.DbSync.Api (
   withDBSyncConnections,
   withScriptConnection,
   withDatumConnection,
+  commitAll,
   mkSyncEnvFromConfig,
   getInsertOptions,
   getTrace,
@@ -44,7 +45,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, newMAChannels, newStakeChannels, useNoCache)
+import Cardano.DbSync.Cache.Types
 import Cardano.DbSync.Config.Cardano
 import Cardano.DbSync.Config.Shelley
 import Cardano.DbSync.Config.Types
@@ -61,14 +62,15 @@ import qualified Cardano.Ledger.Shelley.Genesis as Shelley
 import Cardano.Prelude
 import Cardano.Slotting.Slot (EpochNo (..))
 import Control.Concurrent.Class.MonadSTM.Strict (
+  newEmptyTMVarIO,
   newTBQueueIO,
   newTVarIO,
   readTVar,
   readTVarIO,
+  takeTMVar,
   writeTVar,
  )
-import qualified Control.Concurrent.Class.MonadSTM.Strict.TBQueue as TBQ
-import Control.Concurrent.MVar
+import qualified Control.Concurrent.STM.TBQueue as TBQ
 import Control.Monad.Logger (LoggingT, MonadLoggerIO)
 import Control.Monad.Trans.Resource (MonadUnliftIO)
 import qualified Data.Strict.Maybe as Strict
@@ -247,10 +249,11 @@ hasLedgerState syncEnv =
     NoLedger _ -> False
 
 writePrefetch :: SyncEnv -> CardanoBlock -> IO ()
-writePrefetch syncEnv cblock = do
-  atomically $
-    TBQ.writeTBQueue (pTxInQueue $ envPrefetch syncEnv) $
-      PrefetchTxIdBlock cblock
+writePrefetch _syncEnv _cblock = pure ()
+
+--   atomically $
+--     TBQ.writeTBQueue (pTxInQueue $ envPrefetch syncEnv) $
+--       PrefetchTxIdBlock cblock
 
 mkSyncEnv ::
   Trace IO Text ->
@@ -381,6 +384,18 @@ withGivenConnection toConn syncEnv action = do
   where
     connVar = toConn $ envBackends syncEnv
 
+commitAll :: SyncEnv -> IO ()
+commitAll syncEnv = do
+  maRet <- newEmptyTMVarIO
+  stakeRet <- newEmptyTMVarIO
+  -- queue actions are async here, so we let them run, while blocking on sync actions.
+  atomically $ TBQ.writeTBQueue (macPriorityQueue $ envMAChans syncEnv) $ CommitMA maRet
+  atomically $ TBQ.writeTBQueue (scPriorityQueue $ envStakeChans syncEnv) $ CommitStake stakeRet
+  withScriptConnection syncEnv DB.transactionCommit
+  withDatumConnection syncEnv DB.transactionCommit
+  atomically $ takeTMVar maRet
+  atomically $ takeTMVar stakeRet
+
 mkSyncEnvFromConfig ::
   Trace IO Text ->
   DbConnections ->

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

@@ -45,7 +45,6 @@ import Control.Monad.Logger (LoggingT)
 import qualified Data.ByteString.Short as SBS
 import qualified Data.Strict.Maybe as Strict
 import Database.Persist.Sql
-import Database.Persist.SqlBackend.Internal
 import Ouroboros.Consensus.Byron.Ledger (ByronBlock (..))
 import Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
 import qualified Ouroboros.Consensus.HardFork.Combinator as Consensus
@@ -110,8 +109,12 @@ applyAndInsertBlocks syncEnv firstAfterRollback = go
       let flagList = firstAfterRollback : replicate (length rest) False
       let zippedArgs = zip (DB.BlockKey <$> [newBlockId ..]) flagList
       let (byronBlocks, blocks) = takeWhileByron $ zip zippedArgs (blk : rest)
-      DB.runDbIohkLoggingExceptT backend tracer $ mapM_ (applyAndInsertByronBlock syncEnv) byronBlocks
-      DB.runDbIohkLoggingExceptT backend tracer $ mapM_ (applyAndInsertBlock syncEnv) blocks -- we can use this split to parallelise even further within
+      DB.runDbIohkLoggingExceptT backend tracer $ do
+        mapM_ (applyAndInsertByronBlock syncEnv) byronBlocks
+        liftIO $ commitAll syncEnv
+      DB.runDbIohkLoggingExceptT backend tracer $ do
+        mapM_ (applyAndInsertBlock syncEnv) blocks -- we can use this split to parallelise even further within
+        liftIO $ commitAll syncEnv -- make sure all secondary transactions are commited before the main one.
     backend = envBackend syncEnv
     tracer = getTrace syncEnv
 
@@ -239,6 +242,7 @@ insertBlockRest syncEnv blkNo applyResult tookSnapshot = do
       commited <-
         if withinTwoMin || tookSnapshot
           then do
+            liftIO $ commitAll syncEnv
             lift DB.transactionCommit
             pure True
           else pure False
@@ -247,7 +251,9 @@ insertBlockRest syncEnv blkNo applyResult tookSnapshot = do
         ranIndexes <- liftIO $ getRanIndexes syncEnv
         lift $ addConstraintsIfNotExist syncEnv tracer
         unless ranIndexes $ do
-          lift $ unless commited DB.transactionCommit
+          lift $ unless commited $ do
+            liftIO $ commitAll syncEnv
+            DB.transactionCommit
           liftIO $ runIndexMigrations syncEnv
 
     withinTwoMin = isSyncedWithinSeconds details 120 == SyncFollowing

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

@@ -316,7 +316,7 @@ queryOrInsertSyncMultiAsset syncEnv policy aName = do
     Right maId -> pure maId
     Left _ -> liftIO $ do
       resultVar <- newEmptyTMVarIO
-      atomically $ TBQ.writeTBQueue (macPriorityQueue $ maChan) $ QueryInsertMA policy aName resultVar
+      atomically $ TBQ.writeTBQueue (macPriorityQueue maChan) $ QueryInsertMA policy aName resultVar
       atomically $ takeTMVar resultVar
   where
     maChan = envMAChans syncEnv

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

@@ -272,7 +272,7 @@ data StakeDBAction
   = QueryInsertStake RewAccount CacheAction (StrictTMVar IO DB.StakeAddressId)
   | CacheStake RewAccount DB.StakeAddressId Bool
   | BulkPrefetchStake CardanoBlock
-  | CommitStake
+  | CommitStake (StrictTMVar IO ())
 
 data StakeChannels = StakeChannels
   { scPriorityQueue :: TBQueue StakeDBAction
@@ -293,7 +293,7 @@ data MADBAction
   = QueryInsertMA (PolicyID StandardCrypto) AssetName (StrictTMVar IO DB.MultiAssetId)
   | CacheMA (PolicyID StandardCrypto) AssetName DB.MultiAssetId
   | BulkPrefetchMA CardanoBlock
-  | CommitMA
+  | CommitMA (StrictTMVar IO ())
 
 data MAChannels = MAChannels
   { macPriorityQueue :: TBQueue MADBAction

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

@@ -39,7 +39,9 @@ runMALoop syncEnv =
           liftIO $ atomically $ writeTMVar resVar stakeId
         CacheMA {} -> pure ()
         BulkPrefetchMA _ -> pure ()
-        CommitMA -> DB.transactionCommit
+        CommitMA retVar -> do
+          DB.transactionCommit
+          liftIO $ atomically $ writeTMVar retVar ()
 
     maChan = envMAChans syncEnv
     trce = getTrace syncEnv

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

@@ -37,9 +37,11 @@ runStakeLoop syncEnv =
         QueryInsertStake rewardAcc ca resVar -> do
           stakeId <- resolveInsertRewardAccount syncEnv ca rewardAcc
           liftIO $ atomically $ writeTMVar resVar stakeId
-        CacheStake _ _ _ -> pure ()
+        CacheStake {} -> pure ()
         BulkPrefetchStake _ -> pure ()
-        CommitStake -> DB.transactionCommit
+        CommitStake retVar -> do
+          DB.transactionCommit
+          liftIO $ atomically $ writeTMVar retVar ()
 
     stakeChan = envStakeChans syncEnv
     trce = getTrace syncEnv