change DbAction monad to not have ExceptT

Author: Cmdv

cardano-db-sync/cardano-db-sync.cabal

@@ -188,7 +188,6 @@ library
                       , memory
                       , microlens
                       , monad-control
-                      , monad-logger
                       , network-mux
                       , ouroboros-consensus
                       , ouroboros-consensus-cardano

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

@@ -145,19 +145,19 @@ runDbSync metricsSetters iomgr trce params syncNodeConfigFromFile abortOnPanic =
     void $ unsafeRollback trce (txOutConfigToTableType txOutConfig) pgConfig slotNo
 
   -- This runMigration is ONLY for delayed migrations during sync (like indexes)
-  let runIndexesMigration mode = do
+  let runNearTipMigration mode = do
         msg <- DB.getMaintenancePsqlConf pgConfig
-        logInfo trce $ "Running database migrations in mode " <> textShow mode
+        logInfo trce $ "Running NearTip database migrations in mode " <> textShow mode
         logInfo trce msg
-        when (mode `elem` [DB.Indexes, DB.Full]) $ logWarning trce indexesMsg
+        when (mode `elem` [DB.NearTip, DB.Full]) $ logWarning trce indexesMsg
         DB.runMigrations pgConfig True dbMigrationDir (Just $ DB.LogFileDir "/tmp") mode (txOutConfigToTableType txOutConfig)
 
   runSyncNode
     metricsSetters
     trce
     iomgr
     dbConnectionSetting
-    (void . runIndexesMigration)
+    (void . runNearTipMigration)
     syncNodeConfigFromFile
     params
     syncOpts
@@ -188,7 +188,7 @@ runSyncNode ::
   SyncNodeParams ->
   SyncOptions ->
   IO ()
-runSyncNode metricsSetters trce iomgr dbConnSetting runIndexesMigrationFnc syncNodeConfigFromFile syncNodeParams syncOptions = do
+runSyncNode metricsSetters trce iomgr dbConnSetting runNearTipMigrationFnc syncNodeConfigFromFile syncNodeParams syncOptions = do
   whenJust maybeLedgerDir $
     \enpLedgerStateDir -> do
       createDirectoryIfMissing True (unLedgerStateDir enpLedgerStateDir)
@@ -222,7 +222,7 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runIndexesMigrationFnc syncN
                 genCfg
                 syncNodeConfigFromFile
                 syncNodeParams
-                runIndexesMigrationFnc
+                runNearTipMigrationFnc
 
           -- Warn the user that jsonb datatypes are being removed from the database schema.
           when (isJsonbInSchema && removeJsonbFromSchemaConfig) $ do

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

@@ -14,7 +14,7 @@ module Cardano.DbSync.Api (
   isConsistent,
   getDisableInOutState,
   getRanIndexes,
-  runIndexesMigrations,
+  runNearTipMigrations,
   initPruneConsumeMigration,
   runConsumedTxOutMigrationsMaybe,
   runAddJsonbToSchema,
@@ -120,12 +120,12 @@ getRanIndexes :: SyncEnv -> IO Bool
 getRanIndexes env = do
   readTVarIO $ envIndexes env
 
-runIndexesMigrations :: SyncEnv -> IO ()
-runIndexesMigrations env = do
+runNearTipMigrations :: SyncEnv -> IO ()
+runNearTipMigrations env = do
   haveRan <- readTVarIO $ envIndexes env
   unless haveRan $ do
-    envRunIndexesMigration env DB.Indexes
-    logInfo (getTrace env) "Indexes were created"
+    envRunNearTipMigration env DB.NearTip
+    logInfo (getTrace env) "NearTip migrations were ran successfully."
     atomically $ writeTVar (envIndexes env) True
 
 initPruneConsumeMigration :: Bool -> Bool -> Bool -> Bool -> DB.PruneConsumeMigration
@@ -307,7 +307,7 @@ mkSyncEnvFromConfig ::
   -- | run migration function
   RunMigration ->
   IO (Either SyncNodeError SyncEnv)
-mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runIndexesMigrationFnc =
+mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runNearTipMigrationFnc =
   case genCfg of
     GenesisCardano _ bCfg sCfg _ _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -342,7 +342,7 @@ mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNod
               (SystemStart . Byron.gdStartTime $ Byron.configGenesisData bCfg)
               syncNodeConfigFromFile
               syncNodeParams
-              runIndexesMigrationFnc
+              runNearTipMigrationFnc
 
 mkSyncEnv ::
   Trace IO Text ->
@@ -356,7 +356,7 @@ mkSyncEnv ::
   SyncNodeParams ->
   RunMigration ->
   IO SyncEnv
-mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runIndexesMigrationFnc = do
+mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runNearTipMigrationFnc = do
   dbCNamesVar <- newTVarIO =<< DB.runDbActionIO dbEnv DB.queryRewardAndEpochStakeConstraints
   cache <-
     if soptCache syncOptions
@@ -418,7 +418,7 @@ mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfig
       , envOffChainVoteResultQueue = oarq
       , envOffChainVoteWorkQueue = oawq
       , envOptions = syncOptions
-      , envRunIndexesMigration = runIndexesMigrationFnc
+      , envRunNearTipMigration = runNearTipMigrationFnc
       , envSyncNodeConfig = syncNodeConfigFromFile
       , envSystemStart = systemStart
       }

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

@@ -6,16 +6,6 @@
 
 module Cardano.DbSync.Api.Ledger where
 
-import Control.Concurrent.Class.MonadSTM.Strict (atomically, readTVarIO, writeTVar)
-import Control.Monad.Extra
-import Control.Monad.IO.Class (MonadIO, liftIO)
-import Data.List.Extra
-import Data.Map (Map)
-import qualified Data.Map.Strict as Map
-import qualified Data.Text as Text
-import Lens.Micro
-import Numeric
-
 import Cardano.BM.Trace (logError, logInfo, logWarning)
 import Cardano.Ledger.Allegra.Scripts (Timelock)
 import Cardano.Ledger.Alonzo.Scripts
@@ -26,7 +16,16 @@ import Cardano.Ledger.Core (Value)
 import Cardano.Ledger.Mary.Value
 import Cardano.Ledger.Shelley.LedgerState
 import Cardano.Ledger.TxIn
-import Cardano.Prelude (MonadError (..), textShow)
+import Cardano.Prelude (textShow, throwIO)
+import Control.Concurrent.Class.MonadSTM.Strict (atomically, readTVarIO, writeTVar)
+import Control.Monad.Extra
+import Control.Monad.IO.Class (MonadIO, liftIO)
+import Data.List.Extra
+import Data.Map (Map)
+import qualified Data.Map.Strict as Map
+import qualified Data.Text as Text
+import Lens.Micro
+import Numeric
 import Ouroboros.Consensus.Cardano.Block hiding (CardanoBlock)
 import Ouroboros.Consensus.Ledger.Extended (ExtLedgerState, ledgerState)
 import qualified Ouroboros.Consensus.Shelley.Ledger.Ledger as Consensus
@@ -158,7 +157,7 @@ prepareTxOut syncEnv (TxIn txIntxId (TxIx index), txOut) = do
   let genTxOut = fromTxOut (fromIntegral index) txOut
   eTxId <- queryTxIdWithCache syncEnv txIntxId
   txId <- case eTxId of
-    Left err -> throwError err
+    Left err -> liftIO $ throwIO err
     Right tid -> pure tid
   insertTxOut syncEnv iopts (txId, txHashByteString) genTxOut
   where

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

@@ -57,7 +57,7 @@ data SyncEnv = SyncEnv
   , envOffChainVoteWorkQueue :: !(StrictTBQueue IO OffChainVoteWorkQueue)
   , envOptions :: !SyncOptions
   , envSyncNodeConfig :: !SyncNodeConfig
-  , envRunIndexesMigration :: RunMigration
+  , envRunNearTipMigration :: RunMigration
   , envSystemStart :: !SystemStart
   }
 

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

@@ -1,3 +1,4 @@
+{-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 
@@ -62,7 +63,7 @@ writeEpochBlockDiffToCache ::
   DB.DbAction m ()
 writeEpochBlockDiffToCache cache epCurrent =
   case cache of
-    NoCache -> throwError $ DB.DbError (DB.mkDbCallStack "writeEpochBlockDiffToCache") "Cache is NoCache" Nothing
+    NoCache -> liftIO $ throwIO $ DB.DbError (DB.mkDbCallStack "writeEpochBlockDiffToCache") "Cache is NoCache" Nothing
     ActiveCache ci -> do
       cE <- liftIO $ readTVarIO (cEpoch ci)
       case (ceMapEpoch cE, ceEpochBlockDiff cE) of
@@ -84,12 +85,12 @@ writeToMapEpochCache syncEnv cache latestEpoch = do
           HasLedger hle -> getSecurityParameter $ leProtocolInfo hle
           NoLedger nle -> getSecurityParameter $ nleProtocolInfo nle
   case cache of
-    NoCache -> throwError $ DB.DbError (DB.mkDbCallStack "writeToMapEpochCache") "Cache is NoCache" Nothing
+    NoCache -> liftIO $ throwIO $ DB.DbError (DB.mkDbCallStack "writeToMapEpochCache") "Cache is NoCache" Nothing
     ActiveCache ci -> do
       -- get EpochBlockDiff so we can use the BlockId we stored when inserting blocks
       epochInternalCE <- readEpochBlockDiffFromCache cache
       case epochInternalCE of
-        Nothing -> throwError $ DB.DbError (DB.mkDbCallStack "writeToMapEpochCache") "No epochInternalEpochCache" Nothing
+        Nothing -> liftIO $ throwIO $ DB.DbError (DB.mkDbCallStack "writeToMapEpochCache") "No epochInternalEpochCache" Nothing
         Just ei -> do
           cE <- liftIO $ readTVarIO (cEpoch ci)
           let currentBlockId = ebdBlockId ei

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

@@ -8,6 +8,7 @@ module Cardano.DbSync.Database (
 ) where
 
 import Cardano.BM.Trace (logDebug, logError, logInfo)
+import qualified Cardano.Db as DB
 import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (ConsistentLevel (..), SyncEnv (..))
 import Cardano.DbSync.DbEvent
@@ -19,10 +20,10 @@ import Cardano.DbSync.Rollback
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util
 import Cardano.Prelude hiding (atomically)
-import Cardano.Slotting.Slot (WithOrigin (..))
+import Cardano.Slotting.Slot (SlotNo (..), WithOrigin (..))
 import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Monad.Extra (whenJust)
-import Ouroboros.Network.Block (BlockNo, Point (..))
+import Ouroboros.Network.Block (BlockNo (..), Point (..))
 import Ouroboros.Network.Point (blockPointHash, blockPointSlot)
 
 data NextState
@@ -80,14 +81,16 @@ runDbThread syncEnv metricsSetters queue = do
       logDbState syncEnv
       atomically $ putTMVar resultVar (latestPoints, currentTip)
       processQueue -- Continue processing
-
-    -- Update block and slot height metrics
     updateBlockMetrics :: IO ()
     updateBlockMetrics = do
-      mBlock <- getDbLatestBlockInfo (envDbEnv syncEnv)
-      whenJust mBlock $ \block -> do
-        setDbBlockHeight metricsSetters $ bBlockNo block
-        setDbSlotHeight metricsSetters $ bSlotNo block
+      -- Fire-and-forget async metrics update
+      void $ async $ DB.runPoolDbAction (envDbEnv syncEnv) $ do
+        mBlock <- DB.queryLatestBlock
+        liftIO $ whenJust mBlock $ \block -> do
+          let blockNo = BlockNo $ fromMaybe 0 $ DB.blockBlockNo block
+              slotNo = SlotNo $ fromMaybe 0 $ DB.blockSlotNo block
+          setDbBlockHeight metricsSetters blockNo
+          setDbSlotHeight metricsSetters slotNo
 
 -- | Run the list of 'DbEvent's. Block are applied in a single set (as a transaction)
 -- and other operations are applied one-by-one.

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

@@ -4,14 +4,14 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 {-# OPTIONS_GHC -Wno-unused-matches #-}
 
 module Cardano.DbSync.Default (
   insertListBlocks,
 ) where
 
-import Control.Monad.Logger (LoggingT)
 import qualified Data.ByteString.Short as SBS
 import qualified Data.Set as Set
 import qualified Data.Strict.Maybe as Strict
@@ -50,28 +50,30 @@ insertListBlocks ::
   [CardanoBlock] ->
   IO (Either SyncNodeError ())
 insertListBlocks syncEnv blocks = do
-  result <- DB.runDbIohkLoggingEither tracer (envDbEnv syncEnv) $ do
-    runExceptT $ traverse_ (applyAndInsertBlockMaybe syncEnv tracer) blocks
+  result <-
+    try $
+      DB.runDbIohkLogging tracer (envDbEnv syncEnv) $
+        traverse_ (applyAndInsertBlockMaybe syncEnv tracer) blocks
   case result of
-    Left dbErr -> pure $ Left $ SNErrDatabase dbErr
-    Right (Left syncErr) -> pure $ Left syncErr
-    Right (Right _) -> pure $ Right ()
+    Left (dbErr :: DB.DbError) -> pure $ Left $ SNErrDatabase dbErr
+    Right val -> pure $ Right val
   where
     tracer = getTrace syncEnv
 
 applyAndInsertBlockMaybe ::
+  MonadIO m =>
   SyncEnv ->
   Trace IO Text ->
   CardanoBlock ->
-  ExceptT SyncNodeError (DB.DbAction (LoggingT IO)) ()
+  DB.DbAction m ()
 applyAndInsertBlockMaybe syncEnv tracer cblk = do
   bl <- liftIO $ isConsistent syncEnv
   (!applyRes, !tookSnapshot) <- liftIO (mkApplyResult bl)
   if bl
     then -- In the usual case it will be consistent so we don't need to do any queries. Just insert the block
-      lift $ insertBlock syncEnv cblk applyRes False tookSnapshot
+      insertBlock syncEnv cblk applyRes False tookSnapshot
     else do
-      eiBlockInDbAlreadyId <- lift $ DB.queryBlockIdEither (SBS.fromShort . Consensus.getOneEraHash $ blockHash cblk) ""
+      eiBlockInDbAlreadyId <- DB.queryBlockIdEither (SBS.fromShort . Consensus.getOneEraHash $ blockHash cblk) ""
       -- If the block is already in db, do nothing. If not, delete all blocks with greater 'BlockNo' or
       -- equal, insert the block and restore consistency between ledger and db.
       case eiBlockInDbAlreadyId of
@@ -82,11 +84,11 @@ applyAndInsertBlockMaybe syncEnv tracer cblk = do
               , textShow (getHeaderFields cblk)
               , ". Time to restore consistency."
               ]
-          lift $ rollbackFromBlockNo syncEnv (blockNo cblk)
-          lift $ insertBlock syncEnv cblk applyRes True tookSnapshot
+          rollbackFromBlockNo syncEnv (blockNo cblk)
+          insertBlock syncEnv cblk applyRes True tookSnapshot
           liftIO $ setConsistentLevel syncEnv Consistent
         Right blockId | Just (adaPots, slotNo, epochNo) <- getAdaPots applyRes -> do
-          replaced <- lift $ DB.replaceAdaPots blockId $ mkAdaPots blockId slotNo epochNo adaPots
+          replaced <- DB.replaceAdaPots blockId $ mkAdaPots blockId slotNo epochNo adaPots
           if replaced
             then liftIO $ logInfo tracer $ "Fixed AdaPots for " <> textShow epochNo
             else liftIO $ logInfo tracer $ "Reached " <> textShow epochNo
@@ -114,14 +116,15 @@ applyAndInsertBlockMaybe syncEnv tracer cblk = do
       Generic.neEpoch <$> maybeFromStrict (apNewEpoch appRes)
 
 insertBlock ::
+  MonadIO m =>
   SyncEnv ->
   CardanoBlock ->
   ApplyResult ->
   -- is first Block after rollback
   Bool ->
   -- has snapshot been taken
   Bool ->
-  DB.DbAction (LoggingT IO) ()
+  DB.DbAction m ()
 insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
   !epochEvents <- liftIO $ atomically $ generateNewEpochEvents syncEnv (apSlotDetails applyRes)
   let !applyResult = applyRes {apEvents = sort $ epochEvents <> apEvents applyRes}
@@ -195,7 +198,7 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
       Strict.Nothing | hasLedgerState syncEnv -> Just $ Ledger.Prices minBound minBound
       Strict.Nothing -> Nothing
 
-    commitOrIndexes :: Bool -> Bool -> DB.DbAction (LoggingT IO) ()
+    commitOrIndexes :: MonadIO m => Bool -> Bool -> DB.DbAction m ()
     commitOrIndexes withinTwoMin withinHalfHour = do
       commited <-
         if withinTwoMin || tookSnapshot
@@ -209,7 +212,7 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
         unless ranIndexes $ do
           -- We need to commit the transaction as we are going to run indexes migrations
           DB.commitCurrentTransaction
-          liftIO $ runIndexesMigrations syncEnv
+          liftIO $ runNearTipMigrations syncEnv
 
     blkNo = headerFieldBlockNo $ getHeaderFields cblk
 

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

@@ -22,7 +22,6 @@ import Cardano.DbSync.Types (
 import Cardano.DbSync.Util
 import Cardano.Prelude hiding (from, on, replace)
 import Cardano.Slotting.Slot (unEpochNo)
-import Control.Monad.Logger (LoggingT)
 import Ouroboros.Consensus.Byron.Ledger (ByronBlock (..))
 import Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
 
@@ -33,12 +32,13 @@ import Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
 --    updated on each new block.
 
 epochHandler ::
+  MonadIO m =>
   SyncEnv ->
   Trace IO Text ->
   CacheStatus ->
   Bool ->
   BlockDetails ->
-  DB.DbAction (LoggingT IO) ()
+  DB.DbAction m ()
 epochHandler syncEnv trce cache isNewEpochEvent (BlockDetails cblk details) =
   case cblk of
     BlockByron bblk ->
@@ -57,7 +57,7 @@ epochHandler syncEnv trce cache isNewEpochEvent (BlockDetails cblk details) =
     BlockConway {} -> epochSlotTimecheck
   where
     -- What we do here is completely independent of Shelley/Allegra/Mary eras.
-    epochSlotTimecheck :: DB.DbAction (LoggingT IO) ()
+    epochSlotTimecheck :: MonadIO m => DB.DbAction m ()
     epochSlotTimecheck = do
       when (sdSlotTime details > sdCurrentTime details)
         $ liftIO
@@ -67,12 +67,13 @@ epochHandler syncEnv trce cache isNewEpochEvent (BlockDetails cblk details) =
       updateEpochStart syncEnv cache details isNewEpochEvent False
 
 updateEpochStart ::
+  MonadIO m =>
   SyncEnv ->
   CacheStatus ->
   SlotDetails ->
   Bool ->
   Bool ->
-  DB.DbAction (LoggingT IO) ()
+  DB.DbAction m ()
 updateEpochStart syncEnv cache slotDetails isNewEpochEvent isBoundaryBlock = do
   mLastMapEpochFromCache <- liftIO $ readLastMapEpochFromCache cache
   mEpochBlockDiff <- liftIO $ readEpochBlockDiffFromCache cache
@@ -225,7 +226,7 @@ handleEpochCachingWhenSyncing syncEnv cache newestEpochFromMap epochBlockDiffCac
       newEpoch <- DB.queryCalcEpochEntry $ ebdEpochNo currentEpC
       writeToMapEpochCache syncEnv cache newEpoch
     -- There will always be a EpochBlockDiff at this point in time
-    (_, _) -> throwError $ DB.DbError (DB.mkDbCallStack "handleEpochCachingWhenSyncing") "No caches available to update cache" Nothing
+    (_, _) -> liftIO $ throwIO $ DB.DbError (DB.mkDbCallStack "handleEpochCachingWhenSyncing") "No caches available to update cache" Nothing
 
 -----------------------------------------------------------------------------------------------------
 -- Helper functions