Pipeline stake reward

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

@@ -134,7 +134,7 @@ storePage syncEnv blkEra percQuantum (n, ls) = do
   txOuts <- mapM (prepareTxOut syncEnv blkEra) ls
   txOutIds <- lift $ DB.insertBulkTxOut False $ etoTxOut . fst <$> txOuts
   let maTxOuts = concatMap (mkmaTxOuts txOutVariantType) $ zip txOutIds (snd <$> txOuts)
-  void . lift $ DB.insertBulkMaTxOutPiped [maTxOuts]
+  void . lift $ DB.insertBulkMaTxOutChunked [maTxOuts]
   where
     txOutVariantType = getTxOutVariantType syncEnv
     trce = getTrace syncEnv

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

@@ -18,6 +18,8 @@ module Cardano.DbSync.Cache (
   queryPrevBlockWithCache,
   queryOrInsertStakeAddress,
   queryOrInsertRewardAccount,
+  queryOrInsertStakeAddressBatch,
+  queryPoolKeyOrInsertBatch,
   insertAddressUsingCache,
   insertStakeAddress,
   queryStakeAddrWithCache,
@@ -56,6 +58,8 @@ import qualified Cardano.DbSync.Era.Shelley.Generic.Util as Generic
 import Cardano.DbSync.Era.Shelley.Query
 import Cardano.DbSync.Error (SyncNodeError (..), mkSyncNodeCallStack)
 import Cardano.DbSync.Types
+import qualified Hasql.Pipeline as HsqlP
+import qualified Hasql.Session as HsqlSes
 
 -- Rollbacks make everything harder and the same applies to caching.
 -- After a rollback db entries are deleted, so we need to clean the same
@@ -138,6 +142,65 @@ queryOrInsertStakeAddress ::
 queryOrInsertStakeAddress syncEnv cacheUA nw cred =
   queryOrInsertRewardAccount syncEnv cacheUA $ Ledger.RewardAccount nw cred
 
+-- | Batch version of 'queryOrInsertStakeAddress'. Returns results in the same
+-- order as the input list. Checks cache, pipelines all cache-miss queries,
+-- then inserts any not found. Caller should control chunk size.
+queryOrInsertStakeAddressBatch ::
+  SyncEnv ->
+  CacheAction ->
+  Network ->
+  [StakeCred] ->
+  ExceptT SyncNodeError DB.DbM [DB.StakeAddressId]
+queryOrInsertStakeAddressBatch _syncEnv _cacheUA _nw [] = pure []
+queryOrInsertStakeAddressBatch syncEnv cacheUA nw creds = do
+  -- Check cache for each cred
+  cached <- case envCache syncEnv of
+    NoCache -> pure $ map (const Nothing) creds
+    ActiveCache ci ->
+      withCacheCleanedCheck ci (pure $ map (const Nothing) creds) $ do
+        stakeCache <- liftIO $ readTVarIO (cStake ci)
+        let results = map (\c -> fst <$> queryStakeCache c stakeCache) creds
+        liftIO $ hitCredsN syncEnv (length [() | Just _ <- results])
+        pure results
+  -- Collect cache misses with their cred + serialised bytes
+  let misses = [(c, Ledger.serialiseRewardAccount (mkRA c)) | (Nothing, c) <- zip cached creds]
+  -- Pipeline all cache-miss queries
+  missIds <- pipelineQueryList misses
+  -- Update cache with newly resolved entries
+  case envCache syncEnv of
+    NoCache -> pure ()
+    ActiveCache ci ->
+      liftIO $ atomically $ modifyTVar (cStake ci) $ \sc ->
+        let updateFn = case cacheUA of
+              UpdateCacheStrong -> \acc (c, addrId) -> acc {scStableCache = Map.insert c addrId (scStableCache acc)}
+              UpdateCache -> \acc (c, addrId) -> acc {scLruCache = LRU.insert c addrId (scLruCache acc)}
+              _otherwise -> const
+         in foldl' updateFn sc (zip (map fst misses) missIds)
+  -- Fill in results: cache hits stay, Nothings get filled from missIds in order
+  pure $ fillNothings cached missIds
+  where
+    mkRA = Ledger.RewardAccount nw
+
+    pipelineQueryList :: [(StakeCred, ByteString)] -> ExceptT SyncNodeError DB.DbM [DB.StakeAddressId]
+    pipelineQueryList [] = pure []
+    pipelineQueryList items = do
+      queryResults <- lift $
+        DB.runSession DB.mkDbCallStack $
+          HsqlSes.pipeline $
+            for items $
+              \(_c, bs) -> HsqlP.statement bs DB.queryStakeAddressStmt
+      liftIO $ missCredsN syncEnv (length items)
+      -- For any not found, insert (rare / should not happen for epoch stake)
+      forM (zip items queryResults) $ \((c, bs), mId) -> case mId of
+        Just addrId -> pure addrId
+        Nothing -> insertStakeAddress (mkRA c) (Just bs)
+
+    -- Single-pass merge: replace Nothings with values from the second list
+    fillNothings :: [Maybe a] -> [a] -> [a]
+    fillNothings (Just x : rest) misses = x : fillNothings rest misses
+    fillNothings (Nothing : rest) (x : misses) = x : fillNothings rest misses
+    fillNothings _ _ = []
+
 -- 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 ::
@@ -370,6 +433,56 @@ queryPoolKeyOrInsert syncEnv txt cacheUA logsWarning hsh = do
               ]
       insertPoolKeyWithCache syncEnv cacheUA hsh
 
+-- | Batch version of 'queryPoolKeyOrInsert'. Returns results in the same
+-- order as the input list. Checks cache, pipelines all cache-miss queries,
+-- then inserts any not found. Caller should control chunk size.
+queryPoolKeyOrInsertBatch ::
+  SyncEnv ->
+  CacheAction ->
+  [PoolKeyHash] ->
+  ExceptT SyncNodeError DB.DbM [DB.PoolHashId]
+queryPoolKeyOrInsertBatch _syncEnv _cacheUA [] = pure []
+queryPoolKeyOrInsertBatch syncEnv cacheUA pools = do
+  -- Check cache for each pool
+  cached <- case envCache syncEnv of
+    NoCache -> pure $ map (const Nothing) pools
+    ActiveCache ci -> do
+      mp <- liftIO $ readTVarIO (cPools ci)
+      let results = map (`Map.lookup` mp) pools
+      liftIO $ hitPoolsN syncEnv (length [() | Just _ <- results])
+      pure results
+  -- Collect cache misses
+  let misses = [p | (Nothing, p) <- zip cached pools]
+  -- Pipeline all cache-miss queries
+  missIds <- pipelineQueryList misses
+  -- Update cache with newly resolved entries
+  when (shouldCache cacheUA) $
+    case envCache syncEnv of
+      NoCache -> pure ()
+      ActiveCache ci ->
+        liftIO $ atomically $ modifyTVar (cPools ci) $ \m ->
+          foldl' (\acc (p, phId) -> Map.insert p phId acc) m (zip misses missIds)
+  -- Fill in results
+  pure $ fillNothings cached missIds
+  where
+    pipelineQueryList :: [PoolKeyHash] -> ExceptT SyncNodeError DB.DbM [DB.PoolHashId]
+    pipelineQueryList [] = pure []
+    pipelineQueryList toQuery = do
+      queryResults <- lift $
+        DB.runSession DB.mkDbCallStack $
+          HsqlSes.pipeline $
+            for toQuery $
+              \p -> HsqlP.statement (Generic.unKeyHashRaw p) DB.queryPoolHashIdStmt
+      liftIO $ missPoolsN syncEnv (length toQuery)
+      forM (zip toQuery queryResults) $ \(p, mId) -> case mId of
+        Just phId -> pure phId
+        Nothing -> insertPoolKeyWithCache syncEnv cacheUA p
+
+    fillNothings :: [Maybe a] -> [a] -> [a]
+    fillNothings (Just x : rest) misses = x : fillNothings rest misses
+    fillNothings (Nothing : rest) (x : misses) = x : fillNothings rest misses
+    fillNothings _ _ = []
+
 queryMAWithCache ::
   SyncEnv ->
   PolicyID ->
@@ -548,25 +661,37 @@ withCacheCleanedCheck ci actionIfCleaned actionIfNotCleaned = do
 
 -- Creds
 hitCreds :: SyncEnv -> IO ()
-hitCreds syncEnv =
+hitCreds syncEnv = hitCredsN syncEnv 1
+
+hitCredsN :: SyncEnv -> Int -> IO ()
+hitCredsN syncEnv n =
   atomically $ modifyTVar (envEpochStatistics syncEnv) $ \epochStats ->
-    epochStats {elsCaches = (elsCaches epochStats) {credsHits = 1 + credsHits (elsCaches epochStats), credsQueries = 1 + credsQueries (elsCaches epochStats)}}
+    epochStats {elsCaches = (elsCaches epochStats) {credsHits = fromIntegral n + credsHits (elsCaches epochStats), credsQueries = fromIntegral n + credsQueries (elsCaches epochStats)}}
 
 missCreds :: SyncEnv -> IO ()
-missCreds syncEnv =
+missCreds syncEnv = missCredsN syncEnv 1
+
+missCredsN :: SyncEnv -> Int -> IO ()
+missCredsN syncEnv n =
   atomically $ modifyTVar (envEpochStatistics syncEnv) $ \epochStats ->
-    epochStats {elsCaches = (elsCaches epochStats) {credsQueries = 1 + credsQueries (elsCaches epochStats)}}
+    epochStats {elsCaches = (elsCaches epochStats) {credsQueries = fromIntegral n + credsQueries (elsCaches epochStats)}}
 
 -- Pools
 hitPools :: SyncEnv -> IO ()
-hitPools syncEnv =
+hitPools syncEnv = hitPoolsN syncEnv 1
+
+hitPoolsN :: SyncEnv -> Int -> IO ()
+hitPoolsN syncEnv n =
   atomically $ modifyTVar (envEpochStatistics syncEnv) $ \epochStats ->
-    epochStats {elsCaches = (elsCaches epochStats) {poolsHits = 1 + poolsHits (elsCaches epochStats), poolsQueries = 1 + poolsQueries (elsCaches epochStats)}}
+    epochStats {elsCaches = (elsCaches epochStats) {poolsHits = fromIntegral n + poolsHits (elsCaches epochStats), poolsQueries = fromIntegral n + poolsQueries (elsCaches epochStats)}}
 
 missPools :: SyncEnv -> IO ()
-missPools syncEnv =
+missPools syncEnv = missPoolsN syncEnv 1
+
+missPoolsN :: SyncEnv -> Int -> IO ()
+missPoolsN syncEnv n =
   atomically $ modifyTVar (envEpochStatistics syncEnv) $ \epochStats ->
-    epochStats {elsCaches = (elsCaches epochStats) {poolsQueries = 1 + poolsQueries (elsCaches epochStats)}}
+    epochStats {elsCaches = (elsCaches epochStats) {poolsQueries = fromIntegral n + poolsQueries (elsCaches epochStats)}}
 
 -- Datum
 hitDatum :: SyncEnv -> IO ()

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

@@ -86,7 +86,7 @@ runDbThread syncEnv queue = do
     updateBlockMetrics = do
       let metricsSetters = envMetricSetters syncEnv
       void $ async $ do
-        mBlock <- DB.runDbPoolLogged (fromMaybe mempty $ DB.dbTracer $ envDbEnv syncEnv) (envDbEnv syncEnv) DB.queryLatestBlock
+        mBlock <- DB.runDbPoolTransLogged (fromMaybe mempty $ DB.dbTracer $ envDbEnv syncEnv) (envDbEnv syncEnv) Nothing DB.queryLatestBlock
         liftIO $ whenJust mBlock $ \block -> do
           let blockNo = BlockNo $ fromMaybe 0 $ DB.blockBlockNo block
               slotNo = SlotNo $ fromMaybe 0 $ DB.blockSlotNo block

cardano-db-sync/src/Cardano/DbSync/Era/Byron/Insert.hs

@@ -314,7 +314,7 @@ insertByronTx' syncEnv blkId tx blockIndex = do
   -- Update consumed TxOut records if enabled
   whenConsumeOrPruneTxOut syncEnv $
     lift $
-      DB.updateListTxOutConsumedByTxIdBP [prepUpdate txId <$> resolvedInputs]
+      DB.updateListTxOutConsumedByTxIdChunked [prepUpdate txId <$> resolvedInputs]
 
   -- Return fee amount for caching/epoch calculations
   pure $ unDbLovelace $ vfFee valFee

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

@@ -43,7 +43,7 @@ import Cardano.Slotting.Slot (EpochNo (..), SlotNo)
 import qualified Cardano.Db as DB
 import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (InsertOptions (..), SyncEnv (..))
-import Cardano.DbSync.Cache (queryOrInsertStakeAddress, queryPoolKeyOrInsert)
+import Cardano.DbSync.Cache (queryOrInsertStakeAddressBatch, queryPoolKeyOrInsert, queryPoolKeyOrInsertBatch)
 import Cardano.DbSync.Cache.Types (CacheAction (..))
 import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
 import Cardano.DbSync.Era.Universal.Insert.Certificate (insertPots)
@@ -217,27 +217,23 @@ insertEpochStake ::
   ExceptT SyncNodeError DB.DbM ()
 insertEpochStake syncEnv nw epochNo stakeChunk = do
   DB.ManualDbConstraints {..} <- liftIO $ readTVarIO $ envDbConstraints syncEnv
-  dbStakes <- mapM mkStake stakeChunk
-  let chunckDbStakes = DB.chunkForBulkQuery (Proxy @DB.EpochStake) Nothing dbStakes
-
-  -- minimising the bulk inserts into hundred thousand chunks to improve performance with pipeline
-  lift $ DB.insertBulkEpochStakePiped dbConstraintEpochStake chunckDbStakes
+  let chunks = DB.chunkForBulkQuery (Proxy @DB.EpochStake) Nothing stakeChunk
+  forM_ chunks $ \chunk -> do
+    -- Pipeline stake address and pool hash lookups for this chunk
+    saIds <- queryOrInsertStakeAddressBatch syncEnv UpdateCacheStrong nw (map fst chunk)
+    poolIds <- queryPoolKeyOrInsertBatch syncEnv UpdateCache (map (snd . snd) chunk)
+    let coins = map (fst . snd) chunk
+        dbStakes = zipWith mkStake (zip saIds poolIds) coins
+    lift $ DB.insertBulkEpochStake dbConstraintEpochStake dbStakes
   where
-    mkStake ::
-      (StakeCred, (Shelley.Coin, PoolKeyHash)) ->
-      ExceptT SyncNodeError DB.DbM DB.EpochStake
-    mkStake (saddr, (coin, pool)) = do
-      saId <- queryOrInsertStakeAddress syncEnv UpdateCacheStrong nw saddr
-      poolId <- queryPoolKeyOrInsert syncEnv "insertEpochStake" UpdateCache (ioShelley iopts) pool
-      pure $
-        DB.EpochStake
-          { DB.epochStakeAddrId = saId
-          , DB.epochStakePoolId = poolId
-          , DB.epochStakeAmount = Generic.coinToDbLovelace coin
-          , DB.epochStakeEpochNo = unEpochNo epochNo -- The epoch where this delegation becomes valid.
-          }
-
-    iopts = getInsertOptions syncEnv
+    mkStake :: (DB.StakeAddressId, DB.PoolHashId) -> Shelley.Coin -> DB.EpochStake
+    mkStake (saId, poolId) coin =
+      DB.EpochStake
+        { DB.epochStakeAddrId = saId
+        , DB.epochStakePoolId = poolId
+        , DB.epochStakeAmount = Generic.coinToDbLovelace coin
+        , DB.epochStakeEpochNo = unEpochNo epochNo
+        }
 
 insertRewards ::
   SyncEnv ->
@@ -247,42 +243,26 @@ insertRewards ::
   [(StakeCred, Set Generic.Reward)] ->
   ExceptT SyncNodeError DB.DbM ()
 insertRewards syncEnv nw earnedEpoch spendableEpoch rewardsChunk = do
-  dbRewards <- concatMapM mkRewards rewardsChunk
   DB.ManualDbConstraints {..} <- liftIO $ readTVarIO $ envDbConstraints syncEnv
-  let chunckDbRewards = DB.chunkForBulkQuery (Proxy @DB.Reward) Nothing dbRewards
-  -- minimising the bulk inserts into hundred thousand chunks to improve performance with pipeline
-  lift $ DB.insertBulkRewardsPiped dbConstraintRewards chunckDbRewards
+  let chunks = DB.chunkForBulkQuery (Proxy @DB.Reward) Nothing rewardsChunk
+  forM_ chunks $ \chunk -> do
+    saIds <- queryOrInsertStakeAddressBatch syncEnv UpdateCacheStrong nw (map fst chunk)
+    -- Flatten: expand each (saId, Set Reward) into individual (saId, Reward) pairs
+    let flat = [(saId, rwd) | (saId, (_, rset)) <- zip saIds chunk, rwd <- Set.toList rset]
+    poolIds <- queryPoolKeyOrInsertBatch syncEnv UpdateCache (map (Generic.rewardPool . snd) flat)
+    let dbRewards = zipWith mkReward (zip (map fst flat) poolIds) (map snd flat)
+    lift $ DB.insertBulkRewards dbConstraintRewards dbRewards
   where
-    mkRewards ::
-      (StakeCred, Set Generic.Reward) ->
-      ExceptT SyncNodeError DB.DbM [DB.Reward]
-    mkRewards (saddr, rset) = do
-      saId <- queryOrInsertStakeAddress syncEnv UpdateCacheStrong nw saddr
-      mapM (prepareReward saId) (Set.toList rset)
-
-    prepareReward ::
-      DB.StakeAddressId ->
-      Generic.Reward ->
-      ExceptT SyncNodeError DB.DbM DB.Reward
-    prepareReward saId rwd = do
-      poolId <- queryPool (Generic.rewardPool rwd)
-      pure $
-        DB.Reward
-          { DB.rewardAddrId = saId
-          , DB.rewardType = Generic.rewardSource rwd
-          , DB.rewardAmount = DB.DbLovelace (Generic.rewardAmount rwd)
-          , DB.rewardEarnedEpoch = unEpochNo earnedEpoch
-          , DB.rewardSpendableEpoch = unEpochNo spendableEpoch
-          , DB.rewardPoolId = poolId
-          }
-
-    queryPool ::
-      PoolKeyHash ->
-      ExceptT SyncNodeError DB.DbM DB.PoolHashId
-    queryPool =
-      queryPoolKeyOrInsert syncEnv "insertRewards" UpdateCache (ioShelley iopts)
-
-    iopts = getInsertOptions syncEnv
+    mkReward :: (DB.StakeAddressId, DB.PoolHashId) -> Generic.Reward -> DB.Reward
+    mkReward (saId, poolId) rwd =
+      DB.Reward
+        { DB.rewardAddrId = saId
+        , DB.rewardType = Generic.rewardSource rwd
+        , DB.rewardAmount = DB.DbLovelace (Generic.rewardAmount rwd)
+        , DB.rewardEarnedEpoch = unEpochNo earnedEpoch
+        , DB.rewardSpendableEpoch = unEpochNo spendableEpoch
+        , DB.rewardPoolId = poolId
+        }
 
 insertRewardRests ::
   SyncEnv ->
@@ -292,23 +272,14 @@ insertRewardRests ::
   [(StakeCred, Set Generic.RewardRest)] ->
   ExceptT SyncNodeError DB.DbM ()
 insertRewardRests syncEnv nw earnedEpoch spendableEpoch rewardsChunk = do
-  dbRewards <- concatMapM mkRewards rewardsChunk
-  let chunckDbRewards = DB.chunkForBulkQuery (Proxy @DB.RewardRest) Nothing dbRewards
-  -- minimising the bulk inserts into hundred thousand chunks to improve performance with pipeline
-  lift $ DB.insertBulkRewardRestsPiped chunckDbRewards
+  let chunks = DB.chunkForBulkQuery (Proxy @DB.RewardRest) Nothing rewardsChunk
+  forM_ chunks $ \chunk -> do
+    saIds <- queryOrInsertStakeAddressBatch syncEnv UpdateCacheStrong nw (map fst chunk)
+    let dbRewards = [mkReward saId rwd | (saId, (_, rset)) <- zip saIds chunk, rwd <- Set.toList rset]
+    lift $ DB.insertBulkRewardRests dbRewards
   where
-    mkRewards ::
-      (StakeCred, Set Generic.RewardRest) ->
-      ExceptT SyncNodeError DB.DbM [DB.RewardRest]
-    mkRewards (saddr, rset) = do
-      saId <- queryOrInsertStakeAddress syncEnv UpdateCacheStrong nw saddr
-      pure $ map (prepareReward saId) (Set.toList rset)
-
-    prepareReward ::
-      DB.StakeAddressId ->
-      Generic.RewardRest ->
-      DB.RewardRest
-    prepareReward saId rwd =
+    mkReward :: DB.StakeAddressId -> Generic.RewardRest -> DB.RewardRest
+    mkReward saId rwd =
       DB.RewardRest
         { DB.rewardRestAddrId = saId
         , DB.rewardRestType = Generic.irSource rwd
@@ -325,22 +296,21 @@ insertProposalRefunds ::
   [GovActionRefunded] ->
   ExceptT SyncNodeError DB.DbM ()
 insertProposalRefunds syncEnv nw earnedEpoch spendableEpoch refunds = do
-  dbRewards <- mapM mkReward refunds
-  lift $ DB.insertBulkRewardRests dbRewards
+  let chunks = DB.chunkForBulkQuery (Proxy @DB.RewardRest) Nothing refunds
+  forM_ chunks $ \chunk -> do
+    saIds <- queryOrInsertStakeAddressBatch syncEnv UpdateCacheStrong nw (map (raCredential . garReturnAddr) chunk)
+    let dbRewards = zipWith mkReward saIds chunk
+    lift $ DB.insertBulkRewardRests dbRewards
   where
-    mkReward ::
-      GovActionRefunded ->
-      ExceptT SyncNodeError DB.DbM DB.RewardRest
-    mkReward refund = do
-      saId <- queryOrInsertStakeAddress syncEnv UpdateCacheStrong nw (raCredential $ garReturnAddr refund)
-      pure $
-        DB.RewardRest
-          { DB.rewardRestAddrId = saId
-          , DB.rewardRestType = DB.RwdProposalRefund
-          , DB.rewardRestAmount = Generic.coinToDbLovelace (garDeposit refund)
-          , DB.rewardRestEarnedEpoch = unEpochNo earnedEpoch
-          , DB.rewardRestSpendableEpoch = unEpochNo spendableEpoch
-          }
+    mkReward :: DB.StakeAddressId -> GovActionRefunded -> DB.RewardRest
+    mkReward saId refund =
+      DB.RewardRest
+        { DB.rewardRestAddrId = saId
+        , DB.rewardRestType = DB.RwdProposalRefund
+        , DB.rewardRestAmount = Generic.coinToDbLovelace (garDeposit refund)
+        , DB.rewardRestEarnedEpoch = unEpochNo earnedEpoch
+        , DB.rewardRestSpendableEpoch = unEpochNo spendableEpoch
+        }
 
 insertPoolDepositRefunds ::
   SyncEnv ->