Regroup modules

Author: kderme

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

@@ -54,7 +54,6 @@ library
                         Cardano.DbSync.Config.Node
                         Cardano.DbSync.Config.Shelley
                         Cardano.DbSync.Config.Types
-                        Cardano.DbSync.Database
                         Cardano.DbSync.DbAction
                         Cardano.DbSync.Error
 
@@ -110,14 +109,14 @@ library
 
                         Cardano.DbSync.Metrics
 
+                        Cardano.DbSync.Block
                         Cardano.DbSync.Cache
                         Cardano.DbSync.Cache.Epoch
                         Cardano.DbSync.Cache.FIFO
                         Cardano.DbSync.Cache.LRU
                         Cardano.DbSync.Cache.Stake
                         Cardano.DbSync.Cache.Types
                         Cardano.DbSync.Cache.Util
-                        Cardano.DbSync.Default
                         Cardano.DbSync.Epoch
 
                         Cardano.DbSync.Rollback
@@ -133,7 +132,9 @@ library
                         Cardano.DbSync.LocalStateQuery
                         Cardano.DbSync.StateQuery
                         Cardano.DbSync.Sync
-                        Cardano.DbSync.Threads
+                        Cardano.DbSync.Threads.Database
+                        Cardano.DbSync.Threads.EpochStake
+                        Cardano.DbSync.Threads.Stake
                         Cardano.DbSync.Tracing.ToObjectOrphans
                         Cardano.DbSync.Types
 

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

@@ -33,15 +33,16 @@ import Cardano.DbSync.Api.Types (InsertOptions (..), RunMigration, SyncEnv (..),
 import Cardano.DbSync.Config (configureLogging)
 import Cardano.DbSync.Config.Cardano
 import Cardano.DbSync.Config.Types
-import Cardano.DbSync.Database
 import Cardano.DbSync.DbAction
 import Cardano.DbSync.Era
 import Cardano.DbSync.Error
 import Cardano.DbSync.Ledger.State
 import Cardano.DbSync.OffChain (runFetchOffChainPoolThread, runFetchOffChainVoteThread)
 import Cardano.DbSync.Rollback (unsafeRollback)
 import Cardano.DbSync.Sync (runSyncNodeClient)
-import Cardano.DbSync.Threads
+import Cardano.DbSync.Threads.Database
+import Cardano.DbSync.Threads.EpochStake
+import Cardano.DbSync.Threads.Stake
 import Cardano.DbSync.Tracing.ToObjectOrphans ()
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util.Constraint (queryIsJsonbInSchema)

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

@@ -3,8 +3,6 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TupleSections #-}
-{-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 
 module Cardano.DbSync.Api (
@@ -29,20 +27,12 @@ module Cardano.DbSync.Api (
   getSkipTxIn,
   getPrunes,
   mkSyncEnvFromConfig,
-  verifySnapshotPoint,
   getInsertOptions,
   getTrace,
   getTopLevelConfig,
   getNetwork,
   hasLedgerState,
-  getLatestPoints,
-  getSlotHash,
-  getDbLatestBlockInfo,
-  getDbTipBlockNo,
-  getCurrentTipBlockNo,
   generateNewEpochEvents,
-  logDbState,
-  convertToPoint,
 ) where
 
 import Cardano.BM.Trace (Trace, logInfo, logWarning)
@@ -56,42 +46,34 @@ import Cardano.DbSync.Config.Shelley
 import Cardano.DbSync.Config.Types
 import Cardano.DbSync.Error
 import Cardano.DbSync.Ledger.Event (LedgerEvent (..))
-import Cardano.DbSync.Ledger.State (
-  getHeaderHash,
-  hashToAnnotation,
-  listKnownSnapshots,
-  mkHasLedgerEnv,
- )
-import Cardano.DbSync.Ledger.Types (HasLedgerEnv (..), LedgerStateFile (..), SnapshotPoint (..))
+import Cardano.DbSync.Ledger.State (mkHasLedgerEnv)
+import Cardano.DbSync.Ledger.Types (HasLedgerEnv (..))
 import Cardano.DbSync.LocalStateQuery
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util
 import Cardano.DbSync.Util.Constraint (dbConstraintNamesExists)
 import qualified Cardano.Ledger.BaseTypes as Ledger
 import qualified Cardano.Ledger.Shelley.Genesis as Shelley
 import Cardano.Prelude
-import Cardano.Slotting.Slot (EpochNo (..), SlotNo (..), WithOrigin (..))
+import Cardano.Slotting.Slot (EpochNo (..))
 import Control.Concurrent.Class.MonadSTM.Strict (
   newTBQueueIO,
   newTVarIO,
   readTVar,
   readTVarIO,
   writeTVar,
  )
-import Control.Monad.Trans.Maybe (MaybeT (..))
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (getCurrentTime)
 import Database.Persist.Postgresql (ConnectionString)
 import Database.Persist.Sql (SqlBackend)
-import Ouroboros.Consensus.Block.Abstract (BlockProtocol, HeaderHash, Point (..), fromRawHash)
+import Ouroboros.Consensus.Block.Abstract (BlockProtocol)
 import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
 import Ouroboros.Consensus.Config (SecurityParam (..), TopLevelConfig, configSecurityParam)
 import Ouroboros.Consensus.Node.ProtocolInfo (ProtocolInfo (pInfoConfig))
 import qualified Ouroboros.Consensus.Node.ProtocolInfo as Consensus
 import Ouroboros.Consensus.Protocol.Abstract (ConsensusProtocol)
-import Ouroboros.Network.Block (BlockNo (..), Point (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
-import qualified Ouroboros.Network.Point as Point
 
 setConsistentLevel :: SyncEnv -> ConsistentLevel -> IO ()
 setConsistentLevel env cst = do
@@ -243,60 +225,12 @@ getNetwork sEnv =
 getInsertOptions :: SyncEnv -> InsertOptions
 getInsertOptions = soptInsertOptions . envOptions
 
-getSlotHash :: SqlBackend -> SlotNo -> IO [(SlotNo, ByteString)]
-getSlotHash backend = DB.runDbIohkNoLogging backend . DB.querySlotHash
-
 hasLedgerState :: SyncEnv -> Bool
 hasLedgerState syncEnv =
   case envLedgerEnv syncEnv of
     HasLedger _ -> True
     NoLedger _ -> False
 
-getDbLatestBlockInfo :: SqlBackend -> IO (Maybe TipInfo)
-getDbLatestBlockInfo backend = do
-  runMaybeT $ do
-    block <- MaybeT $ DB.runDbIohkNoLogging backend 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 $
-      TipInfo
-        { bHash = DB.blockHash block
-        , bEpochNo = EpochNo . fromMaybe 0 $ DB.blockEpochNo block
-        , bSlotNo = SlotNo . fromMaybe 0 $ DB.blockSlotNo block
-        , bBlockNo = BlockNo . fromMaybe 0 $ DB.blockBlockNo block
-        }
-
-getDbTipBlockNo :: SyncEnv -> IO (Point.WithOrigin BlockNo)
-getDbTipBlockNo env = do
-  mblk <- getDbLatestBlockInfo (envBackend env)
-  pure $ maybe Point.Origin (Point.At . bBlockNo) mblk
-
-logDbState :: SyncEnv -> IO ()
-logDbState env = do
-  mblk <- getDbLatestBlockInfo (envBackend env)
-  case mblk of
-    Nothing -> logInfo tracer "Database is empty"
-    Just tip -> logInfo tracer $ mconcat ["Database tip is at ", showTip tip]
-  where
-    showTip :: TipInfo -> Text
-    showTip tipInfo =
-      mconcat
-        [ "slot "
-        , textShow (unSlotNo $ bSlotNo tipInfo)
-        , ", block "
-        , textShow (unBlockNo $ bBlockNo tipInfo)
-        ]
-
-    tracer :: Trace IO Text
-    tracer = getTrace env
-
-getCurrentTipBlockNo :: SyncEnv -> IO (WithOrigin BlockNo)
-getCurrentTipBlockNo env = do
-  maybeTip <- getDbLatestBlockInfo (envBackend env)
-  case maybeTip of
-    Just tip -> pure $ At (bBlockNo tip)
-    Nothing -> pure Origin
-
 mkSyncEnv ::
   Trace IO Text ->
   SqlBackend ->
@@ -432,52 +366,6 @@ mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeCon
               syncNodeParams
               runMigrationFnc
 
--- | 'True' is for in memory points and 'False' for on disk
-getLatestPoints :: SyncEnv -> IO [(CardanoPoint, Bool)]
-getLatestPoints env = do
-  case envLedgerEnv env of
-    HasLedger hasLedgerEnv -> do
-      snapshotPoints <- listKnownSnapshots hasLedgerEnv
-      verifySnapshotPoint env snapshotPoints
-    NoLedger _ -> do
-      -- Brings the 5 latest.
-      lastPoints <- DB.runDbIohkNoLogging (envBackend env) DB.queryLatestPoints
-      pure $ mapMaybe convert lastPoints
-  where
-    convert (Nothing, _) = Nothing
-    convert (Just slot, bs) = convertToDiskPoint (SlotNo slot) bs
-
-verifySnapshotPoint :: SyncEnv -> [SnapshotPoint] -> IO [(CardanoPoint, Bool)]
-verifySnapshotPoint env snapPoints =
-  catMaybes <$> mapM validLedgerFileToPoint snapPoints
-  where
-    validLedgerFileToPoint :: SnapshotPoint -> IO (Maybe (CardanoPoint, Bool))
-    validLedgerFileToPoint (OnDisk lsf) = do
-      hashes <- getSlotHash (envBackend env) (lsfSlotNo lsf)
-      let valid = find (\(_, h) -> lsfHash lsf == hashToAnnotation h) hashes
-      case valid of
-        Just (slot, hash) | slot == lsfSlotNo lsf -> pure $ convertToDiskPoint slot hash
-        _ -> pure Nothing
-    validLedgerFileToPoint (InMemory pnt) = do
-      case pnt of
-        GenesisPoint -> pure Nothing
-        BlockPoint slotNo hsh -> do
-          hashes <- getSlotHash (envBackend env) slotNo
-          let valid = find (\(_, dbHash) -> getHeaderHash hsh == dbHash) hashes
-          case valid of
-            Just (dbSlotNo, _) | slotNo == dbSlotNo -> pure $ Just (pnt, True)
-            _ -> pure Nothing
-
-convertToDiskPoint :: SlotNo -> ByteString -> Maybe (CardanoPoint, Bool)
-convertToDiskPoint slot hashBlob = (,False) <$> convertToPoint slot hashBlob
-
-convertToPoint :: SlotNo -> ByteString -> Maybe CardanoPoint
-convertToPoint slot hashBlob =
-  Point . Point.block slot <$> convertHashBlob hashBlob
-  where
-    convertHashBlob :: ByteString -> Maybe (HeaderHash CardanoBlock)
-    convertHashBlob = Just . fromRawHash (Proxy @CardanoBlock)
-
 getSecurityParam :: SyncEnv -> Word64
 getSecurityParam syncEnv =
   case envLedgerEnv syncEnv of

cardano-db-sync/src/Cardano/DbSync/Default.hs renamed to cardano-db-sync/src/Cardano/DbSync/Block.hs

@@ -7,7 +7,7 @@
 {-# LANGUAGE NoImplicitPrelude #-}
 {-# OPTIONS_GHC -Wno-unused-matches #-}
 
-module Cardano.DbSync.Default (
+module Cardano.DbSync.Block (
   insertListBlocks,
 ) where
 
@@ -82,8 +82,8 @@ applyAndInsertBlockMaybe syncEnv tracer cblk = do
               , ". Time to restore consistency."
               ]
           rollbackFromBlockNo syncEnv (blockNo cblk)
-          insertBlock syncEnv cblk applyRes True tookSnapshot
           liftIO $ setConsistentLevel syncEnv Consistent
+          insertBlock syncEnv cblk applyRes True tookSnapshot
         Right blockId | Just (adaPots, slotNo, epochNo) <- getAdaPots applyRes -> do
           replaced <- lift $ DB.replaceAdaPots blockId $ mkAdaPots blockId slotNo epochNo adaPots
           if replaced

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

@@ -29,6 +29,21 @@ import Data.Either.Combinators
 import qualified Data.Map.Strict as Map
 import Database.Persist.Postgresql (SqlBackend)
 
+-- | TO be called only by the stake thread
+resolveInsertRewardAccount ::
+  forall m.
+  (MonadBaseControl IO m, MonadIO m) =>
+  SyncEnv ->
+  CacheAction ->
+  RewAccount ->
+  ReaderT SqlBackend m DB.StakeAddressId
+resolveInsertRewardAccount syncEnv cacheUA ra = do
+  eiStakeId <- queryStakeAddrWithCacheRetBs syncEnv cacheUA False ra -- read only
+  case eiStakeId of
+    Right stakeId -> pure stakeId
+    Left (_, bs) -> insertStakeAddress ra (Just bs)
+
+-- | TO be called only by the stake thread
 -- If the address already exists in the table, it will not be inserted again (due to
 -- the uniqueness constraint) but the function will return the 'StakeAddressId'.
 insertStakeAddress ::

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

@@ -30,7 +30,6 @@ import Cardano.Client.Subscription (subscribe)
 import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (ConsistentLevel (..), LedgerEnv (..), SyncEnv (..), envLedgerEnv, envNetworkMagic, envOptions)
 import Cardano.DbSync.Config
-import Cardano.DbSync.Database
 import Cardano.DbSync.DbAction
 import Cardano.DbSync.LocalStateQuery
 import Cardano.DbSync.Metrics