complete consumedTxOut and start changing tests/smash

Author: Cmdv

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

@@ -54,7 +54,7 @@ library
                         Cardano.DbSync.Config.Shelley
                         Cardano.DbSync.Config.Types
                         Cardano.DbSync.Database
-                        Cardano.DbSync.DbAction
+                        Cardano.DbSync.DbEvent
                         Cardano.DbSync.Error
 
                         Cardano.DbSync.Era

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

@@ -46,7 +46,7 @@ import Cardano.DbSync.Config (configureLogging)
 import Cardano.DbSync.Config.Cardano
 import Cardano.DbSync.Config.Types
 import Cardano.DbSync.Database
-import Cardano.DbSync.DbAction
+import Cardano.DbSync.DbEvent
 import Cardano.DbSync.Era
 import Cardano.DbSync.Error
 import Cardano.DbSync.Ledger.State
@@ -198,7 +198,7 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runMigrationFnc syncNodeConf
           when (not isJsonbInSchema && not removeJsonbFromSchemaConfig) $ do
             liftIO $ logWarning trce "Adding jsonb datatypes back to the database. This can take time."
             liftIO $ runAddJsonbToSchema syncEnv
-          liftIO $ runExtraMigrationsMaybe syncEnv
+          liftIO $ runConsumedTxOutMigrationsMaybe syncEnv
           unless useLedger $ liftIO $ do
             logInfo trce "Migrating to a no ledger schema"
             Db.noLedgerMigrations pool trce

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

@@ -17,7 +17,7 @@ module Cardano.DbSync.Api (
   getRanIndexes,
   runIndexMigrations,
   initPruneConsumeMigration,
-  runExtraMigrationsMaybe,
+  runConsumedTxOutMigrationsMaybe,
   runAddJsonbToSchema,
   runRemoveJsonbFromSchema,
   getSafeBlockNoDiff,
@@ -151,13 +151,13 @@ initPruneConsumeMigration consumed pruneTxOut bootstrap forceTxIn' =
 getPruneConsume :: SyncEnv -> DB.PruneConsumeMigration
 getPruneConsume = soptPruneConsumeMigration . envOptions
 
-runExtraMigrationsMaybe :: SyncEnv -> IO ()
-runExtraMigrationsMaybe syncEnv = do
+runConsumedTxOutMigrationsMaybe :: SyncEnv -> IO ()
+runConsumedTxOutMigrationsMaybe syncEnv = do
   let pcm = getPruneConsume syncEnv
       txOutTableType = getTxOutTableType syncEnv
-  logInfo (getTrace syncEnv) $ "runExtraMigrationsMaybe: " <> textShow pcm
+  logInfo (getTrace syncEnv) $ "runConsumedTxOutMigrationsMaybe: " <> textShow pcm
   DB.runDbIohkNoLogging (envDbEnv syncEnv) $
-    DB.runExtraMigrations
+    DB.runConsumedTxOutMigrations
       (getTrace syncEnv)
       txOutTableType
       (getSafeBlockNoDiff syncEnv)
@@ -170,7 +170,7 @@ runAddJsonbToSchema syncEnv =
 runRemoveJsonbFromSchema ::
   (MonadIO m, AsDbError e) =>
   SyncEnv ->
-  DbAction e m ()
+  DbEvent e m ()
 runRemoveJsonbFromSchema syncEnv = do
   DB.runDbT DB.Write transx
   where

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

@@ -4,17 +4,17 @@
 {-# LANGUAGE NoImplicitPrelude #-}
 
 module Cardano.DbSync.Database (
-  DbAction (..),
+  DbEvent (..),
   ThreadChannels,
-  lengthDbActionQueue,
+  lengthDbEventQueue,
   mkDbApply,
   runDbThread,
 ) where
 
 import Cardano.BM.Trace (logDebug, logError, logInfo)
 import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (ConsistentLevel (..), LedgerEnv (..), SyncEnv (..))
-import Cardano.DbSync.DbAction
+import Cardano.DbSync.DbEvent
 import Cardano.DbSync.Default
 import Cardano.DbSync.Error
 import Cardano.DbSync.Ledger.State
@@ -53,7 +53,7 @@ runDbThread syncEnv metricsSetters queue = do
     -- Main loop to process the queue
     processQueue :: IO ()
     processQueue = do
-      actions <- blockingFlushDbActionQueue queue
+      actions <- blockingFlushDbEventQueue queue
 
       -- Log the number of blocks being processed if there are multiple
       when (length actions > 1) $ do
@@ -65,7 +65,7 @@ runDbThread syncEnv metricsSetters queue = do
         Nothing -> processActions actions
 
     -- Process a list of actions
-    processActions :: [DbAction] -> IO ()
+    processActions :: [DbEvent] -> IO ()
     processActions actions = do
       result <- runExceptT $ runActions syncEnv actions -- runActions is where we start inserting information we recieve from the node.
 
@@ -108,7 +108,7 @@ runDbThread syncEnv metricsSetters queue = do
 --   where
 --     trce = getTrace syncEnv
 --     loop = do
---       xs <- blockingFlushDbActionQueue queue
+--       xs <- blockingFlushDbEventQueue queue
 
 --       when (length xs > 1) $ do
 --         logDebug trce $ "runDbThread: " <> textShow (length xs) <> " blocks"
@@ -136,19 +136,19 @@ runDbThread syncEnv metricsSetters queue = do
 --           atomically $ putTMVar resultVar (latestPoints, currentTip)
 --           loop
 
--- | Run the list of 'DbAction's. Block are applied in a single set (as a transaction)
+-- | Run the list of 'DbEvent's. Block are applied in a single set (as a transaction)
 -- and other operations are applied one-by-one.
 runActions ::
   SyncEnv ->
-  [DbAction] ->
+  [DbEvent] ->
   ExceptT SyncNodeError IO NextState
 runActions syncEnv actions = do
-  dbAction Continue actions
+  dbEvent Continue actions
   where
-    dbAction :: NextState -> [DbAction] -> ExceptT SyncNodeError IO NextState
-    dbAction next [] = pure next
-    dbAction Done _ = pure Done
-    dbAction Continue xs =
+    dbEvent :: NextState -> [DbEvent] -> ExceptT SyncNodeError IO NextState
+    dbEvent next [] = pure next
+    dbEvent Done _ = pure Done
+    dbEvent Continue xs =
       case spanDbApply xs of
         ([], DbFinish : _) -> do
           pure Done
@@ -171,12 +171,12 @@ runActions syncEnv actions = do
               liftIO $ setConsistentLevel syncEnv DBAheadOfLedger
           blockNo <- lift $ getDbTipBlockNo syncEnv
           lift $ atomically $ putTMVar resultVar (points, blockNo)
-          dbAction Continue ys
+          dbEvent Continue ys
         (ys, zs) -> do
           newExceptT $ insertListBlocks syncEnv ys
           if null zs
             then pure Continue
-            else dbAction Continue zs
+            else dbEvent Continue zs
 
 rollbackLedger :: SyncEnv -> CardanoPoint -> IO (Maybe [CardanoPoint])
 rollbackLedger syncEnv point =
@@ -238,14 +238,14 @@ validateConsistentLevel syncEnv stPoint = do
         , show cLevel
         ]
 
--- | Split the DbAction list into a prefix containing blocks to apply and a postfix.
-spanDbApply :: [DbAction] -> ([CardanoBlock], [DbAction])
+-- | Split the DbEvent list into a prefix containing blocks to apply and a postfix.
+spanDbApply :: [DbEvent] -> ([CardanoBlock], [DbEvent])
 spanDbApply lst =
   case lst of
     (DbApplyBlock bt : xs) -> let (ys, zs) = spanDbApply xs in (bt : ys, zs)
     xs -> ([], xs)
 
-hasRestart :: [DbAction] -> Maybe (StrictTMVar IO ([(CardanoPoint, Bool)], WithOrigin BlockNo))
+hasRestart :: [DbEvent] -> Maybe (StrictTMVar IO ([(CardanoPoint, Bool)], WithOrigin BlockNo))
 hasRestart = go
   where
     go [] = Nothing

cardano-db-sync/src/Cardano/DbSync/DbAction.hs renamed to cardano-db-sync/src/Cardano/DbSync/DbEvent.hs

@@ -1,13 +1,13 @@
 {-# LANGUAGE NoImplicitPrelude #-}
 
-module Cardano.DbSync.DbAction (
-  DbAction (..),
+module Cardano.DbSync.DbEvent (
+  DbEvent (..),
   ThreadChannels (..),
-  blockingFlushDbActionQueue,
-  lengthDbActionQueue,
+  blockingFlushDbEventQueue,
+  lengthDbEventQueue,
   mkDbApply,
   newThreadChannels,
-  writeDbActionQueue,
+  writeDbEventQueue,
   waitRollback,
   waitRestartState,
   waitDoneInit,
@@ -23,34 +23,34 @@ import qualified Control.Concurrent.STM.TBQueue as TBQ
 import Ouroboros.Network.Block (BlockNo, Tip (..))
 import qualified Ouroboros.Network.Point as Point
 
-data DbAction
+data DbEvent
   = DbApplyBlock !CardanoBlock
   | DbRollBackToPoint !CardanoPoint !(Tip CardanoBlock) (StrictTMVar IO (Maybe [CardanoPoint], Point.WithOrigin BlockNo))
   | DbRestartState (StrictTMVar IO ([(CardanoPoint, Bool)], Point.WithOrigin BlockNo))
   | DbFinish
 
 data ThreadChannels = ThreadChannels
-  { tcQueue :: TBQueue DbAction
+  { tcQueue :: TBQueue DbEvent
   , tcDoneInit :: !(StrictTVar IO Bool)
   }
 
-mkDbApply :: CardanoBlock -> DbAction
+mkDbApply :: CardanoBlock -> DbEvent
 mkDbApply = DbApplyBlock
 
 -- | This simulates a synhronous operations, since the thread waits for the db
 -- worker thread to finish the rollback.
 waitRollback :: ThreadChannels -> CardanoPoint -> Tip CardanoBlock -> IO (Maybe [CardanoPoint], Point.WithOrigin BlockNo)
 waitRollback tc point serverTip = do
   resultVar <- newEmptyTMVarIO
-  atomically $ writeDbActionQueue tc $ DbRollBackToPoint point serverTip resultVar
+  atomically $ writeDbEventQueue tc $ DbRollBackToPoint point serverTip resultVar
   atomically $ takeTMVar resultVar
 
 waitRestartState :: ThreadChannels -> IO ([(CardanoPoint, Bool)], Point.WithOrigin BlockNo)
 waitRestartState tc = do
   resultVar <- newEmptyTMVarIO
   atomically $ do
     _ <- TBQ.flushTBQueue (tcQueue tc)
-    writeDbActionQueue tc $ DbRestartState resultVar
+    writeDbEventQueue tc $ DbRestartState resultVar
   atomically $ takeTMVar resultVar
 
 waitDoneInit :: ThreadChannels -> IO ()
@@ -68,8 +68,8 @@ runAndSetDone tc action = do
       atomically $ writeTVar (tcDoneInit tc) fl
       pure fl
 
-lengthDbActionQueue :: ThreadChannels -> STM Natural
-lengthDbActionQueue = STM.lengthTBQueue . tcQueue
+lengthDbEventQueue :: ThreadChannels -> STM Natural
+lengthDbEventQueue = STM.lengthTBQueue . tcQueue
 
 newThreadChannels :: IO ThreadChannels
 newThreadChannels =
@@ -81,15 +81,15 @@ newThreadChannels =
     <$> TBQ.newTBQueueIO 47
     <*> newTVarIO False
 
-writeDbActionQueue :: ThreadChannels -> DbAction -> STM ()
-writeDbActionQueue = TBQ.writeTBQueue . tcQueue
+writeDbEventQueue :: ThreadChannels -> DbEvent -> STM ()
+writeDbEventQueue = TBQ.writeTBQueue . tcQueue
 
 -- | Block if the queue is empty and if its not read/flush everything.
 -- Need this because `flushTBQueue` never blocks and we want to block until
 -- there is one item or more.
 -- Use this instead of STM.check to make sure it blocks if the queue is empty.
-blockingFlushDbActionQueue :: ThreadChannels -> IO [DbAction]
-blockingFlushDbActionQueue tc = do
+blockingFlushDbEventQueue :: ThreadChannels -> IO [DbEvent]
+blockingFlushDbEventQueue tc = do
   STM.atomically $ do
     x <- TBQ.readTBQueue $ tcQueue tc
     xs <- TBQ.flushTBQueue $ tcQueue tc

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

@@ -30,7 +30,7 @@ data Metrics = Metrics
   { mNodeBlockHeight :: !Gauge
   -- ^ The block tip number of the remote node.
   , mDbQueueLength :: !Gauge
-  -- ^ The number of @DbAction@ remaining for the database.
+  -- ^ The number of @DbEvent@ remaining for the database.
   , mDbBlockHeight :: !Gauge
   -- ^ The block tip number in the database.
   , mDbSlotHeight :: !Gauge

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

@@ -31,7 +31,7 @@ import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (ConsistentLevel (..), LedgerEnv (..), SyncEnv (..), envLedgerEnv, envNetworkMagic, envOptions)
 import Cardano.DbSync.Config
 import Cardano.DbSync.Database
-import Cardano.DbSync.DbAction
+import Cardano.DbSync.DbEvent
 import Cardano.DbSync.LocalStateQuery
 import Cardano.DbSync.Metrics
 import Cardano.DbSync.Tracing.ToObjectOrphans ()
@@ -221,7 +221,7 @@ dbSyncProtocols syncEnv metricsSetters tc codecConfig version bversion =
               ( chainSyncClientPeerPipelined $
                   chainSyncClient metricsSetters tracer (fst <$> latestPoints) currentTip tc
               )
-          atomically $ writeDbActionQueue tc DbFinish
+          atomically $ writeDbEventQueue tc DbFinish
           -- We should return leftover bytes returned by 'runPipelinedPeer', but
           -- client application do not care about them (it's only important if one
           -- would like to restart a protocol on the same mux and thus bearer).
@@ -350,8 +350,8 @@ chainSyncClient metricsSetters trce latestPoints currentTip tc = do
               setNodeBlockHeight metricsSetters (getTipBlockNo tip)
 
               newSize <- atomically $ do
-                writeDbActionQueue tc $ mkDbApply blk
-                lengthDbActionQueue tc
+                writeDbEventQueue tc $ mkDbApply blk
+                lengthDbEventQueue tc
 
               setDbQueueLength metricsSetters newSize