add custom buffer sizes & additional pipelines

Author: Cmdv

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

@@ -203,16 +203,19 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runNearTipMigrationFnc syncN
     HsqlC.release
     ( \dbConn -> do
         runOrThrowIO $ runExceptT $ do
-          let isLogingEnabled = dncEnableDbLogging syncNodeConfigFromFile
           -- Create connection pool for parallel operations
           pool <- liftIO $ DB.createHasqlConnectionPool [dbConnSetting] 4 -- 4 connections for reasonable parallelism
-          let dbEnv =
-                if isLogingEnabled
-                  then DB.createDbEnv dbConn (Just pool) (Just trce)
-                  else DB.createDbEnv dbConn (Just pool) Nothing
+          let dbEnv = DB.createDbEnv dbConn (Just pool) (Just trce)
           genCfg <- readCardanoGenesisConfig syncNodeConfigFromFile
           isJsonbInSchema <- liftSessionIO mkSyncNodeCallStack $ DB.queryJsonbInSchemaExists dbConn
           logProtocolMagicId trce $ genesisProtocolMagicId genCfg
+          
+          -- Determine the final JSONB state after any schema migrations
+          let finalJsonbInSchema = case (isJsonbInSchema, removeJsonbFromSchemaConfig) of
+                (True, True) -> False   -- Will be removed
+                (False, False) -> True  -- Will be added  
+                (s, _) -> s     -- No change
+          
           syncEnv <-
             ExceptT $
               mkSyncEnvFromConfig
@@ -224,6 +227,7 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runNearTipMigrationFnc syncN
                 syncNodeConfigFromFile
                 syncNodeParams
                 runNearTipMigrationFnc
+                finalJsonbInSchema
 
           -- Warn the user that jsonb datatypes are being removed from the database schema.
           when (isJsonbInSchema && removeJsonbFromSchemaConfig) $ do
@@ -250,8 +254,8 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runNearTipMigrationFnc syncN
               id
               [ runDbThread syncEnv threadChannels
               , runSyncNodeClient metricsSetters syncEnv iomgr trce threadChannels (enpSocketPath syncNodeParams)
-              , runFetchOffChainPoolThread syncEnv syncNodeConfigFromFile
-              , runFetchOffChainVoteThread syncEnv syncNodeConfigFromFile
+              , runFetchOffChainPoolThread syncEnv
+              , runFetchOffChainVoteThread syncEnv
               , runLedgerStateWriteThread (getTrace syncEnv) (envLedgerEnv syncEnv)
               ]
     )

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

@@ -106,7 +106,7 @@ isConsistent env = do
   cst <- getConsistentLevel env
   case cst of
     Consistent -> pure True
-    _otherwise -> pure False
+    _ -> pure False
 
 getDisableInOutState :: SyncEnv -> IO Bool
 getDisableInOutState syncEnv = do
@@ -142,11 +142,13 @@ runConsumedTxOutMigrationsMaybe :: SyncEnv -> IO ()
 runConsumedTxOutMigrationsMaybe syncEnv = do
   let pcm = getPruneConsume syncEnv
       txOutVariantType = getTxOutVariantType syncEnv
+      bulkSize = DB.getTxOutBulkSize txOutVariantType
+
   logInfo (getTrace syncEnv) $ "runConsumedTxOutMigrationsMaybe: " <> textShow pcm
   DB.runDbDirectSilent (envDbEnv syncEnv) $
     DB.runConsumedTxOutMigrations
       (getTrace syncEnv)
-      maxBulkSize
+      bulkSize
       txOutVariantType
       (getSafeBlockNoDiff syncEnv)
       pcm
@@ -308,8 +310,9 @@ mkSyncEnv ::
   SyncNodeConfig ->
   SyncNodeParams ->
   RunMigration ->
+  Bool ->
   IO SyncEnv
-mkSyncEnv metricSetters trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runNearTipMigrationFnc = do
+mkSyncEnv metricSetters trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runNearTipMigrationFnc isJsonbInSchema = do
   dbCNamesVar <- newTVarIO =<< DB.runDbDirectSilent dbEnv DB.queryRewardAndEpochStakeConstraints
   cache <-
     if soptCache syncOptions
@@ -376,6 +379,7 @@ mkSyncEnv metricSetters trce dbEnv syncOptions protoInfo nw nwMagic systemStart
       , envOptions = syncOptions
       , envRunNearTipMigration = runNearTipMigrationFnc
       , envSyncNodeConfig = syncNodeConfigFromFile
+      , envIsJsonbInSchema = isJsonbInSchema
       , envSystemStart = systemStart
       }
   where
@@ -392,8 +396,9 @@ mkSyncEnvFromConfig ::
   SyncNodeParams ->
   -- | run migration function
   RunMigration ->
+  Bool ->
   IO (Either SyncNodeError SyncEnv)
-mkSyncEnvFromConfig metricsSetters trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runNearTipMigrationFnc =
+mkSyncEnvFromConfig metricsSetters trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runNearTipMigrationFnc isJsonbInSchema =
   case genCfg of
     GenesisCardano _ bCfg sCfg _ _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -430,6 +435,7 @@ mkSyncEnvFromConfig metricsSetters trce dbEnv syncOptions genCfg syncNodeConfigF
               syncNodeConfigFromFile
               syncNodeParams
               runNearTipMigrationFnc
+              isJsonbInSchema
 
 -- | 'True' is for in memory points and 'False' for on disk
 getLatestPoints :: SyncEnv -> IO [(CardanoPoint, Bool)]

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

@@ -43,7 +43,6 @@ import Cardano.DbSync.Era.Universal.Insert.Tx (insertTxOut)
 import Cardano.DbSync.Error (SyncNodeError, mkSyncNodeCallStack)
 import Cardano.DbSync.Ledger.State
 import Cardano.DbSync.Types
-import Cardano.DbSync.Util (maxBulkSize)
 
 bootStrapMaybe ::
   SyncEnv ->
@@ -105,12 +104,13 @@ storeUTxO env mp = do
         [ "Inserting "
         , textShow size
         , " tx_out as pages of "
-        , textShow maxBulkSize
+        , textShow bulkSize
         ]
-  mapM_ (storePage env pagePerc) . zip [0 ..] . chunksOf maxBulkSize . Map.toList $ mp
+  mapM_ (storePage env pagePerc) . zip [0 ..] . chunksOf bulkSize . Map.toList $ mp
   where
     trce = getTrace env
-    npages = size `div` maxBulkSize
+    bulkSize = DB.getTxOutBulkSize (getTxOutVariantType env)
+    npages = size `div` bulkSize
     pagePerc :: Float = if npages == 0 then 100.0 else 100.0 / fromIntegral npages
     size = Map.size mp
 

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

@@ -60,6 +60,7 @@ data SyncEnv = SyncEnv
   , envSyncNodeConfig :: !SyncNodeConfig
   , envRunNearTipMigration :: RunMigration
   , envSystemStart :: !SystemStart
+  , envIsJsonbInSchema :: !Bool
   }
 
 data SyncOptions = SyncOptions

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

@@ -89,7 +89,7 @@ cleanCachesForTip cache =
   case cache of
     NoCache -> pure ()
     ActiveCache c ->
-      withCacheCleanedForTipCheck c (pure ()) $
+      withCacheCleanedCheck c (pure ()) $
         liftIO $ do
           -- empty caches not to be used anymore
           atomically $ modifyTVar (cTxIds c) FIFO.cleanupCache
@@ -174,7 +174,7 @@ queryStakeAddrWithCacheRetBs syncEnv cacheUA ra@(Ledger.RewardAccount _ cred) =
   case envCache syncEnv of
     NoCache -> (,bs) <$> resolveStakeAddress bs
     ActiveCache ci -> do
-      result <- withCacheCleanedForTipCheck ci (resolveStakeAddress bs) $ do
+      result <- withCacheCleanedCheck ci (resolveStakeAddress bs) $ do
         stakeCache <- liftIO $ readTVarIO (cStake ci)
         case queryStakeCache cred stakeCache of
           Just (addrId, stakeCache') -> do
@@ -379,7 +379,7 @@ queryMAWithCache syncEnv policyId asset =
   case envCache syncEnv of
     NoCache -> lift queryDb
     ActiveCache ci -> do
-      withCacheCleanedForTipCheck ci (lift queryDb) $ do
+      withCacheCleanedCheck ci (lift queryDb) $ do
         mp <- liftIO $ readTVarIO (cMultiAssets ci)
         case LRU.lookup (policyId, asset) mp of
           Just (maId, mp') -> do
@@ -437,7 +437,7 @@ queryTxIdWithCache syncEnv txIdLedger = do
     -- Direct database query if no cache.
     NoCache -> lift qTxHash
     ActiveCache ci ->
-      withCacheCleanedForTipCheck ci (lift qTxHash) $ do
+      withCacheCleanedCheck ci (lift qTxHash) $ do
         -- Read current cache state.
         cacheTx <- liftIO $ readTVarIO (cTxIds ci)
 
@@ -485,7 +485,7 @@ insertBlockAndCache syncEnv block =
   case envCache syncEnv of
     NoCache -> lift insBlck
     ActiveCache ci ->
-      withCacheCleanedForTipCheck ci (lift insBlck) $ do
+      withCacheCleanedCheck ci (lift insBlck) $ do
         bid <- lift insBlck
         liftIO $ do
           missPrevBlock syncEnv
@@ -502,7 +502,7 @@ queryDatum syncEnv hsh = do
   case envCache syncEnv of
     NoCache -> lift queryDtm
     ActiveCache ci -> do
-      withCacheCleanedForTipCheck ci (lift queryDtm) $ do
+      withCacheCleanedCheck ci (lift queryDtm) $ do
         mp <- liftIO $ readTVarIO (cDatum ci)
         case LRU.lookup hsh mp of
           Just (datumId, mp') -> do
@@ -527,24 +527,24 @@ insertDatumAndCache cache hsh dt = do
   case cache of
     NoCache -> pure datumId
     ActiveCache ci ->
-      withCacheCleanedForTipCheck ci (pure datumId) $ do
+      withCacheCleanedCheck ci (pure datumId) $ do
         liftIO $
           atomically $
             modifyTVar (cDatum ci) $
               LRU.insert hsh datumId
         pure datumId
 
-withCacheCleanedForTipCheck ::
+withCacheCleanedCheck ::
   MonadIO m =>
   CacheInternal ->
   m a -> -- Action to perform if cache is cleaned for tip
   m a -> -- Action to perform if cache is not cleaned for tip
   m a
-withCacheCleanedForTipCheck ci ifCleanedForTip ifNotCleanedForTip = do
+withCacheCleanedCheck ci actionIfCleaned actionIfNotCleaned = do
   isCacheCleanedForTip <- liftIO $ readTVarIO (cIsCacheCleanedForTip ci)
   if isCacheCleanedForTip
-    then ifCleanedForTip
-    else ifNotCleanedForTip
+    then actionIfCleaned
+    else actionIfNotCleaned
 
 -- Creds
 hitCreds :: SyncEnv -> IO ()

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

@@ -111,7 +111,8 @@ data CacheCapacity = CacheCapacity
   , cacheCapacityDatum :: !Word64
   , cacheCapacityMultiAsset :: !Word64
   , cacheCapacityTx :: !Word64
-  , -- Optimisation target sizes for Map-based caches (used every 100k blocks)
+  , -- Used by optimiseCaches function to trim Map-based caches that can grow without bounds,
+    -- unlike LRU caches which have built-in capacity limits. Trimming keeps most recent entries.
     cacheOptimisePools :: !Word64
   , cacheOptimiseStake :: !Word64
   }
@@ -145,16 +146,16 @@ textShowCacheStats stats (ActiveCache ic) = do
   mAssets <- readTVarIO (cMultiAssets ic)
   txIds <- readTVarIO (cTxIds ic)
   address <- readTVarIO (cAddress ic)
-  pure $
-    mconcat
+  pure
+    $ mconcat
       [ "\n\nEpoch Cache Statistics: "
       , "\n  Caches Cleaned For Tip: " <> textShow isCacheCleanedForTip
       , textCacheSection "  Stake Addresses" (scLruCache stakeHashRaws) (scStableCache stakeHashRaws) (credsHits stats) (credsQueries stats)
       , textMapSection "  Pools" pools (poolsHits stats) (poolsQueries stats)
       , textLruSection "  Datums" datums (datumHits stats) (datumQueries stats)
       , textLruSection "  Addresses" address (addressHits stats) (addressQueries stats)
       , textLruSection "  Multi Assets" mAssets (multiAssetsHits stats) (multiAssetsQueries stats)
-      , textPrevBlockSection
+      , textPrevBlockSection "  Previous Block"
       , textFifoSection "  TxId" txIds (txIdsHits stats) (txIdsQueries stats)
       ]
   where
@@ -196,9 +197,9 @@ textShowCacheStats stats (ActiveCache ic) = do
         , hitMissStats hits queries
         ]
 
-    textPrevBlockSection =
+    textPrevBlockSection title =
       mconcat
-        [ "\nPrevious Block: "
+        [ "\n" <> title <> ": "
         , hitMissStats (prevBlockHits stats) (prevBlockQueries stats)
         ]
 
@@ -231,8 +232,9 @@ newEmptyCache CacheCapacity {..} = liftIO $ do
   cEpoch <- newTVarIO initCacheEpoch
   cTxIds <- newTVarIO (FIFO.empty cacheCapacityTx)
 
-  pure . ActiveCache $
-    CacheInternal
+  pure
+    . ActiveCache
+    $ CacheInternal
       { cIsCacheCleanedForTip = cIsCacheCleanedForTip
       , cStake = cStake
       , cPools = cPools

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

@@ -66,7 +66,6 @@ coalesceConfig pcfg ncfg adjustGenesisPath = do
       , dncProtocol = ncProtocol ncfg
       , dncRequiresNetworkMagic = ncRequiresNetworkMagic ncfg
       , dncEnableLogging = pcEnableLogging pcfg
-      , dncEnableDbLogging = pcEnableDbLogging pcfg
       , dncEnableMetrics = pcEnableMetrics pcfg
       , dncPrometheusPort = pcPrometheusPort pcfg
       , dncPBftSignatureThreshold = ncPBftSignatureThreshold ncfg

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

@@ -1,6 +1,5 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 
 module Cardano.DbSync.Config.Byron (

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

@@ -127,7 +127,6 @@ data SyncNodeConfig = SyncNodeConfig
   , dncProtocol :: !SyncProtocol
   , dncRequiresNetworkMagic :: !RequiresNetworkMagic
   , dncEnableLogging :: !Bool
-  , dncEnableDbLogging :: !Bool
   , dncEnableMetrics :: !Bool
   , dncPrometheusPort :: !Int
   , dncPBftSignatureThreshold :: !(Maybe Double)
@@ -156,7 +155,6 @@ data SyncPreConfig = SyncPreConfig
   , pcNodeConfigFile :: !NodeConfigFile
   , pcEnableFutureGenesis :: !Bool
   , pcEnableLogging :: !Bool
-  , pcEnableDbLogging :: !Bool
   , pcEnableMetrics :: !Bool
   , pcPrometheusPort :: !Int
   , pcInsertConfig :: !SyncInsertConfig
@@ -405,7 +403,6 @@ parseGenSyncNodeConfig o =
     <*> fmap NodeConfigFile (o .: "NodeConfigFile")
     <*> fmap (fromMaybe True) (o .:? "EnableFutureGenesis")
     <*> o .: "EnableLogging"
-    <*> fmap (fromMaybe False) (o .:? "EnableDbLogging")
     <*> o .: "EnableLogMetrics"
     <*> fmap (fromMaybe 8080) (o .:? "PrometheusPort")
     <*> o .:? "insert_options" .!= def