Add Resolve Inputs thread

Author: kderme

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

@@ -135,6 +135,7 @@ library
                         Cardano.DbSync.Threads.Database
                         Cardano.DbSync.Threads.EpochStake
                         Cardano.DbSync.Threads.Stake
+                        Cardano.DbSync.Threads.TxInResolve
                         Cardano.DbSync.Tracing.ToObjectOrphans
                         Cardano.DbSync.Types
 

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

@@ -32,6 +32,7 @@ module Cardano.DbSync.Api (
   getTopLevelConfig,
   getNetwork,
   hasLedgerState,
+  writePrefetch,
   generateNewEpochEvents,
 ) where
 
@@ -63,6 +64,7 @@ import Control.Concurrent.Class.MonadSTM.Strict (
   readTVarIO,
   writeTVar,
  )
+import qualified Control.Concurrent.Class.MonadSTM.Strict.TBQueue as TBQ
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (getCurrentTime)
 import Database.Persist.Postgresql (ConnectionString)
@@ -231,6 +233,12 @@ hasLedgerState syncEnv =
     HasLedger _ -> True
     NoLedger _ -> False
 
+writePrefetch :: SyncEnv -> CardanoBlock -> IO ()
+writePrefetch syncEnv cblock = do
+  atomically $
+    TBQ.writeTBQueue (pTxInQueue $ envPrefetch syncEnv) $
+      PrefetchTxIdBlock cblock
+
 mkSyncEnv ::
   Trace IO Text ->
   SqlBackend ->
@@ -258,6 +266,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
             , cacheCapacityTx = 100000
             }
       else pure useNoCache
+  prefetch <- newPrefetch
   consistentLevelVar <- newTVarIO Unchecked
   indexesVar <- newTVarIO $ enpForceIndexes syncNP
   bts <- getBootstrapInProgress trce (isTxOutConsumedBootstrap' syncNodeConfigFromFile) backend
@@ -295,6 +304,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
       { envBackend = backend
       , envBootstrap = bootstrapVar
       , envCache = cache
+      , envPrefetch = prefetch
       , envConnectionString = connectionString
       , envConsistentLevel = consistentLevelVar
       , envDbConstraints = dbCNamesVar

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

@@ -1,6 +1,5 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE NoImplicitPrelude #-}
 
 module Cardano.DbSync.Api.Types (
   SyncEnv (..),
@@ -10,34 +9,39 @@ module Cardano.DbSync.Api.Types (
   RunMigration,
   ConsistentLevel (..),
   CurrentEpochNo (..),
+  Prefetch (..),
+  PrefetchTxId (..),
+  newPrefetch,
 ) where
 
 import qualified Cardano.Db as DB
 import Cardano.DbSync.Cache.Types (CacheStatus, StakeChannels)
 import Cardano.DbSync.Config.Types (SyncNodeConfig)
+import qualified Cardano.DbSync.Era.Shelley.Generic.Tx.Types as Generic
 import Cardano.DbSync.Ledger.Types (HasLedgerEnv)
 import Cardano.DbSync.LocalStateQuery (NoLedgerEnv)
 import Cardano.DbSync.Types (
+  CardanoBlock,
   OffChainPoolResult,
   OffChainPoolWorkQueue,
   OffChainVoteResult,
   OffChainVoteWorkQueue,
  )
-import Cardano.Prelude (Bool, Eq, IO, Show, Word64)
 import Cardano.Slotting.Slot (EpochNo (..))
-import Control.Concurrent.Class.MonadSTM.Strict (
-  StrictTVar,
- )
+import Control.Concurrent.Class.MonadSTM.Strict (StrictTVar, newTBQueueIO, newTVarIO)
 import Control.Concurrent.Class.MonadSTM.Strict.TBQueue (StrictTBQueue)
+import Data.Map (Map)
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (UTCTime)
+import Data.Word (Word64)
 import Database.Persist.Postgresql (ConnectionString)
 import Database.Persist.Sql (SqlBackend)
 import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
 
 data SyncEnv = SyncEnv
   { envBackend :: !SqlBackend
+  , envPrefetch :: !Prefetch
   , envCache :: !CacheStatus
   , envConnectionString :: !ConnectionString
   , envConsistentLevel :: !(StrictTVar IO ConsistentLevel)
@@ -101,3 +105,16 @@ data ConsistentLevel = Consistent | DBAheadOfLedger | Unchecked
 newtype CurrentEpochNo = CurrentEpochNo
   { cenEpochNo :: Strict.Maybe EpochNo
   }
+
+data PrefetchTxId = PrefetchTxIdBlock CardanoBlock | PrefetchTxIdBlocks [CardanoBlock]
+
+data Prefetch = Prefetch
+  { pTxInQueue :: StrictTBQueue IO PrefetchTxId
+  , pTxIn :: StrictTVar IO (Map Generic.TxInKey (Maybe (DB.TxId, Either Generic.TxInKey DB.TxOutIdW, Maybe DB.DbLovelace)))
+  }
+
+newPrefetch :: IO Prefetch
+newPrefetch =
+  Prefetch
+    <$> newTBQueueIO 1000
+    <*> newTVarIO mempty

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

@@ -64,6 +64,7 @@ applyAndInsertBlockMaybe ::
   ExceptT SyncNodeError (ReaderT SqlBackend (LoggingT IO)) ()
 applyAndInsertBlockMaybe syncEnv tracer cblk = do
   bl <- liftIO $ isConsistent syncEnv
+  when bl $ liftIO $ writePrefetch syncEnv cblk
   (!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
@@ -83,6 +84,7 @@ applyAndInsertBlockMaybe syncEnv tracer cblk = do
               ]
           rollbackFromBlockNo syncEnv (blockNo cblk)
           liftIO $ setConsistentLevel syncEnv Consistent
+          when bl $ liftIO $ writePrefetch syncEnv cblk
           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

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Generic/Tx/Shelley.hs

@@ -12,6 +12,7 @@ module Cardano.DbSync.Era.Shelley.Generic.Tx.Shelley (
   getTxSize,
   mkTxIn,
   fromTxIn,
+  toTxInKey,
   mkTxOut,
   mkTxWithdrawals,
   mkTxWithdrawal,
@@ -123,12 +124,14 @@ mkTxOut txBody = zipWith fromTxOut [0 ..] $ toList (txBody ^. Core.outputsTxBody
         , txOutDatum = NoDatum -- Shelley does not support plutus data
         }
 
+toTxInKey :: Ledger.TxIn StandardCrypto -> TxInKey
+toTxInKey (Ledger.TxIn txId (TxIx w64)) = TxInKey txId w64
+
 fromTxIn :: Ledger.TxIn StandardCrypto -> TxIn
-fromTxIn (Ledger.TxIn (Ledger.TxId txid) (TxIx w64)) =
+fromTxIn txIn =
   TxIn
-    { txInIndex = w64
+    { txInKey = toTxInKey txIn
     , txInRedeemerIndex = Nothing
-    , txInTxId = Ledger.TxId txid
     }
 
 txHashId :: (EraCrypto era ~ StandardCrypto, Core.EraTx era) => Core.Tx era -> ByteString

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Generic/Tx/Types.hs

@@ -12,6 +12,7 @@ module Cardano.DbSync.Era.Shelley.Generic.Tx.Types (
   TxCertificate (..),
   TxWithdrawal (..),
   TxIn (..),
+  TxInKey (..),
   TxOut (..),
   TxRedeemer (..),
   TxScript (..),
@@ -97,9 +98,14 @@ data TxWithdrawal = TxWithdrawal
   , txwAmount :: !Coin
   }
 
+data TxInKey = TxInKey
+  { txInTxId :: !(Ledger.TxId StandardCrypto)
+  , txInIndex :: !Word64
+  }
+  deriving (Eq, Show, Ord)
+
 data TxIn = TxIn
-  { txInIndex :: !Word64
-  , txInTxId :: !(Ledger.TxId StandardCrypto)
+  { txInKey :: !TxInKey
   , txInRedeemerIndex :: !(Maybe Word64) -- This only has a meaning for Alonzo.
   }
   deriving (Show)
@@ -174,7 +180,7 @@ getMaybeDatumHash (Just hsh) = DatumHash hsh
 sumTxOutCoin :: [TxOut] -> Coin
 sumTxOutCoin = Coin . sum . map (unCoin . txOutAdaValue)
 
-toTxHash :: TxIn -> ByteString
+toTxHash :: TxInKey -> ByteString
 toTxHash = unTxHash . txInTxId
 
 class AlonzoEraTxBody era => DBScriptPurpose era where

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Query.hs

@@ -6,7 +6,6 @@
 module Cardano.DbSync.Era.Shelley.Query (
   resolveStakeAddress,
   resolveInputTxOutId,
-  resolveInputValue,
   resolveInputTxOutIdValue,
   queryResolveInputCredentials,
 ) where
@@ -26,18 +25,14 @@ import Database.Esqueleto.Experimental (
 resolveStakeAddress :: MonadIO m => ByteString -> ReaderT SqlBackend m (Either LookupFail StakeAddressId)
 resolveStakeAddress addr = queryStakeAddress addr renderByteArray
 
-resolveInputTxOutId :: MonadIO m => SyncEnv -> Generic.TxIn -> ReaderT SqlBackend m (Either LookupFail (TxId, TxOutIdW))
+resolveInputTxOutId :: MonadIO m => SyncEnv -> Generic.TxInKey -> ReaderT SqlBackend m (Either LookupFail (TxId, TxOutIdW))
 resolveInputTxOutId syncEnv txIn =
   queryTxOutId (getTxOutTableType syncEnv) (Generic.toTxHash txIn, fromIntegral (Generic.txInIndex txIn))
 
-resolveInputValue :: MonadIO m => SyncEnv -> Generic.TxIn -> ReaderT SqlBackend m (Either LookupFail (TxId, DbLovelace))
-resolveInputValue syncEnv txIn =
-  queryTxOutValue (getTxOutTableType syncEnv) (Generic.toTxHash txIn, fromIntegral (Generic.txInIndex txIn))
-
-resolveInputTxOutIdValue :: MonadIO m => SyncEnv -> Generic.TxIn -> ReaderT SqlBackend m (Either LookupFail (TxId, TxOutIdW, DbLovelace))
+resolveInputTxOutIdValue :: MonadIO m => SyncEnv -> Generic.TxInKey -> ReaderT SqlBackend m (Either LookupFail (TxId, TxOutIdW, DbLovelace))
 resolveInputTxOutIdValue syncEnv txIn =
   queryTxOutIdValue (getTxOutTableType syncEnv) (Generic.toTxHash txIn, fromIntegral (Generic.txInIndex txIn))
 
-queryResolveInputCredentials :: MonadIO m => SyncEnv -> Generic.TxIn -> ReaderT SqlBackend m (Either LookupFail (Maybe ByteString, Bool))
+queryResolveInputCredentials :: MonadIO m => SyncEnv -> Generic.TxInKey -> ReaderT SqlBackend m (Either LookupFail (Maybe ByteString, Bool))
 queryResolveInputCredentials syncEnv txIn = do
   queryTxOutCredentials (getTxOutTableType syncEnv) (Generic.toTxHash txIn, fromIntegral (Generic.txInIndex txIn))