bulk consummed txout

Author: Cmdv

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

@@ -1,5 +1,6 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE NoImplicitPrelude #-}
 
 module Cardano.DbSync.Era.Universal.Insert.Grouped (
@@ -17,7 +18,7 @@ module Cardano.DbSync.Era.Universal.Insert.Grouped (
 import qualified Data.List as List
 import qualified Data.Text as Text
 
-import Cardano.BM.Trace (Trace, logWarning)
+import Cardano.BM.Trace (logWarning)
 import Cardano.Db (DbLovelace (..), MinIds (..))
 import qualified Cardano.Db as DB
 import qualified Cardano.Db.Schema.Variants.TxOutAddress as VA
@@ -26,6 +27,7 @@ import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (InsertOptions (..), SyncEnv (..), SyncOptions (..))
 import Cardano.DbSync.Cache (queryTxIdWithCacheEither)
 import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
+import Cardano.DbSync.Era.Shelley.Generic.Util (unTxHash)
 import Cardano.DbSync.Era.Shelley.Query
 import Cardano.Prelude
 
@@ -98,9 +100,18 @@ insertBlockGroupedData syncEnv grouped = do
       then pure []
       else DB.insertBulkTxIn $ etiTxIn <$> groupedTxIn grouped
   whenConsumeOrPruneTxOut syncEnv $ do
+    -- Resolve remaining inputs
     etis <- resolveRemainingInputs (groupedTxIn grouped) $ zip txOutIds (fst <$> groupedTxOut grouped)
-    updateTuples <- mapM (prepareUpdates tracer) etis
-    DB.updateListTxOutConsumedByTxId $ catMaybes updateTuples
+    -- Categorise resolved inputs for bulk vs individual processing
+    let (hashBasedUpdates, idBasedUpdates, failedInputs) = categorizeResolvedInputs etis
+    -- Bulk process hash-based updates
+    unless (null hashBasedUpdates) $ DB.updateConsumedByTxHashBulk txOutVariantType hashBasedUpdates
+    -- Individual process ID-based updates
+    unless (null idBasedUpdates) $ do
+      DB.updateListTxOutConsumedByTxId idBasedUpdates
+    -- Log failures
+    mapM_ (liftIO . logWarning tracer . ("Failed to find output for " <>) . Text.pack . show) failedInputs
+
   void . DB.insertBulkTxMetadata removeJsonbFromSchema $ groupedTxMetadata grouped
   void . DB.insertBulkMaTxMint $ groupedTxMint grouped
   pure $ makeMinId txInIds txOutIds maTxOutIds
@@ -109,6 +120,24 @@ insertBlockGroupedData syncEnv grouped = do
     txOutVariantType = getTxOutVariantType syncEnv
     removeJsonbFromSchema = ioRemoveJsonbFromSchema $ soptInsertOptions $ envOptions syncEnv
 
+    categorizeResolvedInputs :: [ExtendedTxIn] -> ([DB.BulkConsumedByHash], [(DB.TxOutIdW, DB.TxId)], [ExtendedTxIn])
+    categorizeResolvedInputs etis =
+      let (hashBased, idBased, failed) = foldr categorizeOne ([], [], []) etis
+       in (hashBased, idBased, failed)
+      where
+        categorizeOne ExtendedTxIn {..} (hAcc, iAcc, fAcc) =
+          case etiTxOutId of
+            Right txOutId ->
+              (hAcc, (txOutId, DB.txInTxInId etiTxIn) : iAcc, fAcc)
+            Left genericTxIn ->
+              let bulkData =
+                    DB.BulkConsumedByHash
+                      { bchTxHash = unTxHash (Generic.txInTxId genericTxIn)
+                      , bchOutputIndex = Generic.txInIndex genericTxIn
+                      , bchConsumingTxId = DB.txInTxInId etiTxIn
+                      }
+               in (bulkData : hAcc, iAcc, fAcc)
+
     makeMinId :: [DB.TxInId] -> [DB.TxOutIdW] -> [DB.MaTxOutIdW] -> DB.MinIdsWrapper
     makeMinId txInIds txOutIds maTxOutIds =
       case txOutVariantType of
@@ -148,17 +177,6 @@ mkmaTxOuts _txOutVariantType (txOutId, mmtos) = mkmaTxOut <$> mmtos
               , VA.maTxOutAddressTxOutId = txOutId'
               }
 
-prepareUpdates ::
-  MonadIO m =>
-  Trace IO Text ->
-  ExtendedTxIn ->
-  m (Maybe (DB.TxOutIdW, DB.TxId))
-prepareUpdates trce eti = case etiTxOutId eti of
-  Right txOutId -> pure $ Just (txOutId, DB.txInTxInId (etiTxIn eti))
-  Left _ -> do
-    liftIO $ logWarning trce $ "Failed to find output for " <> Text.pack (show eti)
-    pure Nothing
-
 insertReverseIndex ::
   MonadIO m =>
   DB.BlockId ->

cardano-db/src/Cardano/Db/Statement/ConsumedTxOut.hs

@@ -10,7 +10,7 @@
 module Cardano.Db.Statement.ConsumedTxOut where
 
 import Cardano.BM.Trace (Trace, logInfo)
-import Cardano.Prelude (Int64, textShow)
+import Cardano.Prelude (ByteString, Int64, textShow)
 import Contravariant.Extras (contrazip2, contrazip3)
 import Control.Exception (throwIO)
 import Control.Monad (unless, when)
@@ -567,36 +567,68 @@ deletePageEntries txOutVariantType entries =
 
 --------------------------------------------------------------------------------
 
--- Statement for updating TxOut entries with consumed_by_tx_id
-updatePageEntriesStmt ::
+-- | Data for bulk consumption using tx hash
+data BulkConsumedByHash = BulkConsumedByHash
+  { bchTxHash :: !ByteString
+  , bchOutputIndex :: !Word64
+  , bchConsumingTxId :: !Id.TxId
+  }
+
+-- | Bulk update consumed_by_tx_id using tx hash + index
+updateConsumedByTxHashBulk ::
+  MonadIO m =>
+  TxOutVariantType ->
+  [BulkConsumedByHash] ->
+  DbAction m ()
+updateConsumedByTxHashBulk txOutVariantType consumedData =
+  unless (null consumedData) $ do
+    let dbCallStack = mkDbCallStack "updateConsumedByTxHashBulk"
+    case txOutVariantType of
+      TxOutVariantCore ->
+        runDbSession dbCallStack $
+          HsqlSes.statement consumedData (updateConsumedByTxHashBulkStmt @SVC.TxOutCore)
+      TxOutVariantAddress ->
+        runDbSession dbCallStack $
+          HsqlSes.statement consumedData (updateConsumedByTxHashBulkStmt @SVA.TxOutAddress)
+
+updateConsumedByTxHashBulkStmt ::
   forall a.
   (DbInfo a) =>
-  HsqlStmt.Statement [ConsumedTriplet] ()
-updatePageEntriesStmt =
+  HsqlStmt.Statement [BulkConsumedByHash] ()
+updateConsumedByTxHashBulkStmt =
   HsqlStmt.Statement sql encoder HsqlD.noResult True
   where
     tableN = tableName (Proxy @a)
     sql =
       TextEnc.encodeUtf8 $
         Text.concat
-          [ "WITH entries AS ("
-          , "  SELECT unnest($1::bigint[]) as tx_out_tx_id,"
-          , "         unnest($2::int[]) as tx_out_index,"
-          , "         unnest($3::bigint[]) as tx_in_tx_id"
+          [ "WITH consumption_data AS ("
+          , "  SELECT unnest($1::bytea[]) as tx_hash,"
+          , "         unnest($2::bigint[]) as output_index,"
+          , "         unnest($3::bigint[]) as consuming_tx_id"
           , ")"
           , "UPDATE " <> tableN
-          , "SET consumed_by_tx_id = entries.tx_in_tx_id"
-          , "WHERE (tx_id, index) IN (SELECT tx_out_tx_id, tx_out_index FROM entries)"
+          , "SET consumed_by_tx_id = consumption_data.consuming_tx_id"
+          , "FROM consumption_data"
+          , "INNER JOIN tx ON tx.hash = consumption_data.tx_hash"
+          , "WHERE " <> tableN <> ".tx_id = tx.id"
+          , "  AND " <> tableN <> ".index = consumption_data.output_index"
           ]
+    encoder = contramap extractBulkData bulkConsumedByHashEncoder
 
-    encoder = contramap extract encodeConsumedTripletBulk
+extractBulkData :: [BulkConsumedByHash] -> ([ByteString], [Word64], [Id.TxId])
+extractBulkData xs =
+  ( map bchTxHash xs
+  , map bchOutputIndex xs
+  , map bchConsumingTxId xs
+  )
 
-    extract :: [ConsumedTriplet] -> ([Id.TxId], [Word64], [Id.TxId])
-    extract xs =
-      ( map ctTxOutTxId xs
-      , map ctTxOutIndex xs
-      , map ctTxInTxId xs
-      )
+bulkConsumedByHashEncoder :: HsqlE.Params ([ByteString], [Word64], [Id.TxId])
+bulkConsumedByHashEncoder =
+  contrazip3
+    (bulkEncoder $ HsqlE.nonNullable HsqlE.bytea)
+    (bulkEncoder $ HsqlE.nonNullable $ fromIntegral >$< HsqlE.int8)
+    (bulkEncoder $ HsqlE.nonNullable $ Id.getTxId >$< HsqlE.int8)
 
 --------------------------------------------------------------------------------
 

cardano-db/test/Test/IO/Cardano/Db/Rollback.hs

@@ -134,7 +134,7 @@ createAndInsertBlocks blockCount =
 
           txIds <- mapM insertTx (mkTxs blkId 8)
           let txId = case txIds of
-                (x:_) -> x
+                (x : _) -> x
                 [] -> error "mkTxs returned empty list" -- This shouldn't happen with mkTxs blkId 8
           void $ insertTxIn (TxIn txId txOutId 0 Nothing)
           void $ insertTxOut (mkTxOutCore blkId txId)