fix restart and transactions

Author: Cmdv

cardano-chain-gen/test/Test/Cardano/Db/Mock/Validate.hs

@@ -66,8 +66,8 @@ import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
 import Cardano.DbSync.Era.Shelley.Generic.Util
 import qualified Cardano.Ledger.Address as Ledger
 import Cardano.Ledger.BaseTypes
-import Cardano.Ledger.Shelley.LedgerState (EraCertState)
 import qualified Cardano.Ledger.Core as Core
+import Cardano.Ledger.Shelley.LedgerState (EraCertState)
 import Cardano.Mock.Forging.Tx.Generic
 import Cardano.Mock.Forging.Types
 import Cardano.Prelude (MonadIO)

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

@@ -25,6 +25,7 @@ module Cardano.DbSync (
   extractSyncOptions,
 ) where
 
+import Control.Concurrent.Async
 import Control.Monad.Extra (whenJust)
 import qualified Data.Strict.Maybe as Strict
 import qualified Data.Text as Text
@@ -36,7 +37,6 @@ import Ouroboros.Network.NodeToClient (IOManager, withIOManager)
 import Paths_cardano_db_sync (version)
 import System.Directory (createDirectoryIfMissing)
 import Prelude (id)
-import Control.Concurrent.Async
 
 import Cardano.BM.Trace (Trace, logError, logInfo, logWarning)
 import qualified Cardano.Crypto as Crypto
@@ -240,14 +240,18 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runDelayedMigrationFnc syncN
 
           -- communication channel between datalayer thread and chainsync-client thread
           threadChannels <- liftIO newThreadChannels
-          liftIO $ race_
-            (runDbThread syncEnv metricsSetters threadChannels)  -- Main App thread
-            (mapConcurrently_ id [                               -- Non-critical threads
-              runSyncNodeClient metricsSetters syncEnv iomgr trce threadChannels (enpSocketPath syncNodeParams),
-              runFetchOffChainPoolThread syncEnv syncNodeConfigFromFile,
-              runFetchOffChainVoteThread syncEnv syncNodeConfigFromFile,
-              runLedgerStateWriteThread (getTrace syncEnv) (envLedgerEnv syncEnv)
-            ])
+          liftIO $
+            race_
+              (runDbThread syncEnv metricsSetters threadChannels) -- Main App thread
+              ( mapConcurrently_
+                  id
+                  [ -- Non-critical threads
+                    runSyncNodeClient metricsSetters syncEnv iomgr trce threadChannels (enpSocketPath syncNodeParams)
+                  , runFetchOffChainPoolThread syncEnv syncNodeConfigFromFile
+                  , runFetchOffChainVoteThread syncEnv syncNodeConfigFromFile
+                  , runLedgerStateWriteThread (getTrace syncEnv) (envLedgerEnv syncEnv)
+                  ]
+              )
     )
   where
     useShelleyInit :: SyncNodeConfig -> Bool

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

@@ -27,7 +27,7 @@ import Cardano.Ledger.Core (Value)
 import Cardano.Ledger.Mary.Value
 import Cardano.Ledger.Shelley.LedgerState
 import Cardano.Ledger.TxIn
-import Cardano.Prelude (textShow, MonadError (..))
+import Cardano.Prelude (MonadError (..), textShow)
 import Control.Concurrent.Class.MonadSTM.Strict (atomically, readTVarIO, writeTVar)
 import Control.Monad.Extra
 import Control.Monad.IO.Class (MonadIO, liftIO)

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

@@ -242,7 +242,7 @@ queryPoolKeyWithCache cache cacheUA hsh =
           liftIO $ missPools (cStats ci)
           mPhId <- DB.queryPoolHashId (Generic.unKeyHashRaw hsh)
           case mPhId of
-            Nothing -> throwError $ DB.DbError (DB.mkDbCallStack "queryPoolKeyWithCache") "ActiveCache queryPoolHashId" Nothing
+            Nothing -> pure $ Left $ DB.DbError (DB.mkDbCallStack "queryPoolKeyWithCache") "ActiveCache queryPoolHashId" Nothing
             Just phId -> do
               -- missed so we can't evict even with 'EvictAndReturn'
               when (shouldCache cacheUA) $
@@ -412,7 +412,7 @@ queryMAWithCache cache policyId asset =
 queryTxIdWithCacheEither ::
   MonadIO m =>
   CacheStatus ->
-  Ledger.TxId ->  -- Use the original input type
+  Ledger.TxId -> -- Use the original input type
   DB.DbAction m (Either DB.DbError DB.TxId)
 queryTxIdWithCacheEither cache txIdLedger = do
   case cache of
@@ -441,7 +441,7 @@ queryTxIdWithCacheEither cache txIdLedger = do
               -- Return lookup failure.
               Left err -> pure $ Left err
   where
-    txHash = Generic.unTxHash txIdLedger  -- Convert to ByteString for DB query
+    txHash = Generic.unTxHash txIdLedger -- Convert to ByteString for DB query
     qTxHash = do
       result <- DB.queryTxId txHash
       case result of
@@ -514,8 +514,13 @@ queryTxIdWithCache cache txIdLedger = do
       result <- DB.queryTxId txHash
       case result of
         Just txId -> pure $ Right txId
-        Nothing -> pure $ Left $ DB.DbError (DB.mkDbCallStack "queryTxIdWithCacheEither")
-                          ("TxId not found for hash: " <> textShow txHash) Nothing
+        Nothing ->
+          pure $
+            Left $
+              DB.DbError
+                (DB.mkDbCallStack "queryTxIdWithCacheEither")
+                ("TxId not found for hash: " <> textShow txHash)
+                Nothing
 
 tryUpdateCacheTx ::
   MonadIO m =>

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

@@ -152,6 +152,7 @@ textShowStats (ActiveCache ic) = do
       , textLruSection "Multi Assets" mAssets (multiAssetsHits stats) (multiAssetsQueries stats)
       , textPrevBlockSection stats
       , textFifoSection "TxId" txIds (txIdsHits stats) (txIdsQueries stats)
+      , "\n-----------------------------------------------------------------"
       ]
   where
     textCacheSection title cacheLru cacheStable hits queries =

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

@@ -19,7 +19,7 @@ import Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
 import qualified Ouroboros.Consensus.HardFork.Combinator as Consensus
 import Ouroboros.Network.Block (blockHash, blockNo, getHeaderFields, headerFieldBlockNo, unBlockNo)
 
-import Cardano.BM.Trace (logInfo, Trace)
+import Cardano.BM.Trace (Trace, logInfo)
 import qualified Cardano.Db as DB
 import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Ledger
@@ -76,11 +76,12 @@ applyAndInsertBlockMaybe syncEnv tracer cblk = do
       -- equal, insert the block and restore consistency between ledger and db.
       case eiBlockInDbAlreadyId of
         Left _ -> do
-          liftIO . logInfo tracer $ mconcat
-            [ "Received block which is not in the db with "
-            , textShow (getHeaderFields cblk)
-            , ". Time to restore consistency."
-            ]
+          liftIO . logInfo tracer $
+            mconcat
+              [ "Received block which is not in the db with "
+              , textShow (getHeaderFields cblk)
+              , ". Time to restore consistency."
+              ]
           lift $ rollbackFromBlockNo syncEnv (blockNo cblk)
           lift $ insertBlock syncEnv cblk applyRes True tookSnapshot
           liftIO $ setConsistentLevel syncEnv Consistent
@@ -198,7 +199,7 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
     commitOrIndexes withinTwoMin withinHalfHour = do
       commited <-
         if withinTwoMin || tookSnapshot
-          then do pure True
+          then pure True
           else pure False
       when withinHalfHour $ do
         bootStrapMaybe syncEnv

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

@@ -279,7 +279,6 @@ insertByronTx syncEnv blkId tx blockIndex = do
   where
     iopts = getInsertOptions syncEnv
 
-
 insertByronTx' ::
   (MonadIO m) =>
   SyncEnv ->
@@ -289,7 +288,11 @@ insertByronTx' ::
   DB.DbAction m Word64
 insertByronTx' syncEnv blkId tx blockIndex = do
   -- Resolve all transaction inputs - any failure will throw via MonadError
-  resolvedInputs <- mapM (resolveTxInputsByron txOutVariantType) (toList $ Byron.txInputs (Byron.taTx tx))
+  resolvedResults <- mapM (resolveTxInputsByron txOutVariantType) (toList $ Byron.txInputs (Byron.taTx tx))
+
+  resolvedInputs <- case sequence resolvedResults of
+    Right inputs -> pure inputs
+    Left dbErr -> throwError dbErr
 
   -- Calculate transaction fee
   valFee <- case calculateTxFee (Byron.taTx tx) resolvedInputs of
@@ -435,14 +438,17 @@ insertTxIn _tracer txInTxId (Byron.TxInUtxo _txHash inIndex, txOutTxId, _, _) =
 
 -------------------------------------------------------------------------------
 
-resolveTxInputsByron :: (MonadIO m) => DB.TxOutVariantType -> Byron.TxIn -> DB.DbAction m (Byron.TxIn, DB.TxId, DB.TxOutIdW, DbLovelace)
+resolveTxInputsByron ::
+  (MonadIO m) =>
+  DB.TxOutVariantType ->
+  Byron.TxIn ->
+  DB.DbAction m (Either DB.DbError (Byron.TxIn, DB.TxId, DB.TxOutIdW, DbLovelace))
 resolveTxInputsByron txOutVariantType txIn@(Byron.TxInUtxo txHash index) = do
   result <- DB.queryTxOutIdValueEither txOutVariantType (Byron.unTxHash txHash, fromIntegral index)
-  case result of
-    Right res -> pure $ convert res
-    Left dbErr -> throwError dbErr  -- Use MonadError instead of Either
+  pure $ case result of
+    Right res -> Right $ convert res
+    Left dbErr -> Left dbErr -- Return Either instead of throwing
   where
-    convert :: (DB.TxId, DB.TxOutIdW, DbLovelace) -> (Byron.TxIn, DB.TxId, DB.TxOutIdW, DbLovelace)
     convert (txId, txOutId, lovelace) = (txIn, txId, txOutId, lovelace)
 
 calculateTxFee :: Byron.Tx -> [(Byron.TxIn, DB.TxId, DB.TxOutIdW, DbLovelace)] -> Either SyncNodeError ValueFee

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

@@ -152,8 +152,8 @@ insertValidateShelleyGenesisDist syncEnv networkName cfg shelleyInitiation = do
         , DB.blockSize = 0
         , DB.blockTime = configStartTime cfg
         , DB.blockTxCount = expectedTxCount
-        -- Genesis block does not have a protocol version, so set this to '0'.
-        , DB.blockProtoMajor = 0
+        , -- Genesis block does not have a protocol version, so set this to '0'.
+          DB.blockProtoMajor = 0
         , DB.blockProtoMinor = 0
         , DB.blockVrfKey = Nothing
         , DB.blockOpCert = Nothing
@@ -180,8 +180,9 @@ validateGenesisDistribution syncEnv prunes networkName cfg bid expectedTxCount =
   metaMaybe <- DB.queryMeta
   meta <- case metaMaybe of
     Just m -> pure m
-    Nothing -> throwError $
-      DB.DbError dbCallStack "Meta table is empty during validation - this should not happen" Nothing
+    Nothing ->
+      throwError $
+        DB.DbError dbCallStack "Meta table is empty during validation - this should not happen" Nothing
 
   when (DB.metaStartTime meta /= configStartTime cfg) $
     throwError $

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

@@ -204,8 +204,11 @@ resolveTxInputs syncEnv hasConsumed needsValue groupedOutputs txIn = do
       case (resolveInMemory txIn groupedOutputs, hasConsumed, needsValue) of
         (Nothing, _, _) ->
           -- Only throw if in-memory resolution also fails
-          throwError $ DB.DbError (DB.mkDbCallStack "resolveTxInputs")
-            ("TxOut not found for TxIn: " <> textShow txIn) Nothing
+          throwError $
+            DB.DbError
+              (DB.mkDbCallStack "resolveTxInputs")
+              ("TxOut not found for TxIn: " <> textShow txIn)
+              Nothing
         (Just eutxo, True, True) ->
           pure $ convertFoundValue (etoTxOut eutxo)
         (Just eutxo, _, _) ->

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

@@ -243,7 +243,7 @@ runFetchOffChainPoolThread syncEnv syncNodeConfigFromFile = do
           tDelay
           -- load the offChain vote work queue using the db
           _ <-
-            runExceptT $ DB.runDbIohkLoggingEither trce dbEnv $
+            DB.runDbIohkLoggingEither trce dbEnv $
               loadOffChainPoolWorkQueue trce (envOffChainPoolWorkQueue threadSyncEnv)
           poolq <- atomically $ flushTBQueue (envOffChainPoolWorkQueue threadSyncEnv)
           manager <- Http.newManager tlsManagerSettings
@@ -280,7 +280,7 @@ runFetchOffChainVoteThread syncEnv syncNodeConfigFromFile = do
             tDelay
             -- load the offChain vote work queue using the db
             _ <-
-              runExceptT $ DB.runDbIohkLoggingEither trce dbEnv $
+              DB.runDbIohkLoggingEither trce dbEnv $
                 loadOffChainVoteWorkQueue trce (envOffChainVoteWorkQueue threadSyncEnv)
             voteq <- atomically $ flushTBQueue (envOffChainVoteWorkQueue threadSyncEnv)
             now <- liftIO Time.getPOSIXTime