fix index creation when close to tip

Author: Cmdv

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

@@ -145,7 +145,7 @@ runDbSync metricsSetters iomgr trce params syncNodeConfigFromFile abortOnPanic =
     void $ unsafeRollback trce (txOutConfigToTableType txOutConfig) pgConfig slotNo
 
   -- This runMigration is ONLY for delayed migrations during sync (like indexes)
-  let runDelayedMigration mode = do
+  let runIndexesMigration mode = do
         msg <- DB.getMaintenancePsqlConf pgConfig
         logInfo trce $ "Running database migrations in mode " <> textShow mode
         logInfo trce msg
@@ -157,7 +157,7 @@ runDbSync metricsSetters iomgr trce params syncNodeConfigFromFile abortOnPanic =
     trce
     iomgr
     dbConnectionSetting
-    (void . runDelayedMigration)
+    (void . runIndexesMigration)
     syncNodeConfigFromFile
     params
     syncOpts
@@ -188,7 +188,7 @@ runSyncNode ::
   SyncNodeParams ->
   SyncOptions ->
   IO ()
-runSyncNode metricsSetters trce iomgr dbConnSetting runDelayedMigrationFnc syncNodeConfigFromFile syncNodeParams syncOptions = do
+runSyncNode metricsSetters trce iomgr dbConnSetting runIndexesMigrationFnc syncNodeConfigFromFile syncNodeParams syncOptions = do
   whenJust maybeLedgerDir $
     \enpLedgerStateDir -> do
       createDirectoryIfMissing True (unLedgerStateDir enpLedgerStateDir)
@@ -220,7 +220,7 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runDelayedMigrationFnc syncN
                 genCfg
                 syncNodeConfigFromFile
                 syncNodeParams
-                runDelayedMigrationFnc
+                runIndexesMigrationFnc
 
           -- Warn the user that jsonb datatypes are being removed from the database schema.
           when (isJsonbInSchema && removeJsonbFromSchemaConfig) $ do

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

@@ -15,7 +15,7 @@ module Cardano.DbSync.Api (
   getIsConsumedFixed,
   getDisableInOutState,
   getRanIndexes,
-  runIndexMigrations,
+  runIndexesMigrations,
   initPruneConsumeMigration,
   runConsumedTxOutMigrationsMaybe,
   runAddJsonbToSchema,
@@ -131,11 +131,11 @@ getRanIndexes :: SyncEnv -> IO Bool
 getRanIndexes env = do
   readTVarIO $ envIndexes env
 
-runIndexMigrations :: SyncEnv -> IO ()
-runIndexMigrations env = do
+runIndexesMigrations :: SyncEnv -> IO ()
+runIndexesMigrations env = do
   haveRan <- readTVarIO $ envIndexes env
   unless haveRan $ do
-    envRunDelayedMigration env DB.Indexes
+    envRunIndexesMigration env DB.Indexes
     logInfo (getTrace env) "Indexes were created"
     atomically $ writeTVar (envIndexes env) True
 
@@ -318,7 +318,7 @@ mkSyncEnvFromConfig ::
   -- | run migration function
   RunMigration ->
   IO (Either SyncNodeError SyncEnv)
-mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runDelayedMigrationFnc =
+mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runIndexesMigrationFnc =
   case genCfg of
     GenesisCardano _ bCfg sCfg _ _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -353,7 +353,7 @@ mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNod
               (SystemStart . Byron.gdStartTime $ Byron.configGenesisData bCfg)
               syncNodeConfigFromFile
               syncNodeParams
-              runDelayedMigrationFnc
+              runIndexesMigrationFnc
 
 mkSyncEnv ::
   Trace IO Text ->
@@ -367,7 +367,7 @@ mkSyncEnv ::
   SyncNodeParams ->
   RunMigration ->
   IO SyncEnv
-mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runDelayedMigrationFnc = do
+mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runIndexesMigrationFnc = do
   dbCNamesVar <- newTVarIO =<< DB.runDbActionIO dbEnv DB.queryRewardAndEpochStakeConstraints
   cache <-
     if soptCache syncOptions
@@ -429,7 +429,7 @@ mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfig
       , envOffChainVoteResultQueue = oarq
       , envOffChainVoteWorkQueue = oawq
       , envOptions = syncOptions
-      , envRunDelayedMigration = runDelayedMigrationFnc
+      , envRunIndexesMigration = runIndexesMigrationFnc
       , envSyncNodeConfig = syncNodeConfigFromFile
       , envSystemStart = systemStart
       }

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

@@ -57,7 +57,7 @@ data SyncEnv = SyncEnv
   , envOffChainVoteWorkQueue :: !(StrictTBQueue IO OffChainVoteWorkQueue)
   , envOptions :: !SyncOptions
   , envSyncNodeConfig :: !SyncNodeConfig
-  , envRunDelayedMigration :: RunMigration
+  , envRunIndexesMigration :: RunMigration
   , envSystemStart :: !SystemStart
   }
 

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

@@ -205,9 +205,11 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
         bootStrapMaybe syncEnv
         ranIndexes <- liftIO $ getRanIndexes syncEnv
         addConstraintsIfNotExist syncEnv tracer
-        unless ranIndexes $
-          liftIO $
-            runIndexMigrations syncEnv
+
+        unless ranIndexes $ do
+          -- We need to commit the transaction as we are going to run indexes migrations
+          DB.commitCurrentTransaction
+          liftIO $ runIndexesMigrations syncEnv
 
     blkNo = headerFieldBlockNo $ getHeaderFields cblk
 

cardano-db/cardano-db.cabal

@@ -30,6 +30,7 @@ library
                         -Wunused-packages
 
   exposed-modules:      Cardano.Db
+                        Cardano.Db.Progress
                         Cardano.Db.Schema.Core
                         Cardano.Db.Schema.Variants
                         Cardano.Db.Schema.Variants.TxOutAddress
@@ -85,7 +86,6 @@ library
                       , cardano-crypto-class
                       , cardano-ledger-core
                       , cardano-prelude
-                      , conduit-extra
                       , containers
                       , contra-tracer
                       , contravariant-extras

cardano-db/src/Cardano/Db/Migration.hs

@@ -25,15 +25,12 @@ module Cardano.Db.Migration (
 ) where
 
 import Cardano.Prelude (textShow)
-import Control.Exception (Exception, SomeException, handle)
+import Control.Exception (Exception)
 import Control.Monad.Extra
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Control.Monad.Logger (NoLoggingT)
-import Control.Monad.Trans.Resource (runResourceT)
 import qualified Data.ByteString.Char8 as BS
 import Data.Char (isDigit)
-import Data.Conduit.Binary (sinkHandle)
-import Data.Conduit.Process (sourceCmdWithConsumer)
 import Data.Either (partitionEithers)
 import Data.List ((\\))
 import qualified Data.List as List
@@ -53,7 +50,6 @@ import System.FilePath (takeExtension, takeFileName, (</>))
 import System.IO (
   Handle,
   IOMode (AppendMode),
-  hFlush,
   hPrint,
   hPutStrLn,
   stdout,
@@ -63,13 +59,14 @@ import Text.Read (readMaybe)
 
 import Cardano.BM.Trace (Trace)
 import Cardano.Crypto.Hash (Blake2b_256, ByteString, Hash, hashToStringAsHex, hashWith)
-import Cardano.Db.Migration.Haskell
 import Cardano.Db.Migration.Version
 import Cardano.Db.PGConfig
 import Cardano.Db.Run
 import Cardano.Db.Schema.Variants (TxOutVariantType (..))
 import qualified Cardano.Db.Statement.Function.Core as DB
 import qualified Cardano.Db.Types as DB
+import System.Process (readProcessWithExitCode)
+import Cardano.Db.Progress (withProgress, updateProgress)
 
 newtype MigrationDir
   = MigrationDir FilePath
@@ -104,19 +101,32 @@ runMigrations pgconfig quiet migrationDir mLogfiledir mToRun txOutVariantType =
     (_, []) ->
       error $ "Empty schema dir " ++ show migrationDir
     (Nothing, scripts) -> do
-      -- Remove the pattern match that separates first script
       putStrLn "Running:"
-      (scripts', ranAll) <- filterMigrations scripts -- Filter ALL scripts including first
-      forM_ scripts' $ applyMigration' Nothing stdout
+      (scripts', ranAll) <- filterMigrations scripts
+
+      -- Replace just this forM_ with progress bar
+      withProgress (length scripts') "Database migrations" $ \progressRef -> do
+        forM_ (zip [1 :: Integer ..] scripts') $ \(i, script) -> do
+          updateProgress progressRef (fromIntegral i) $
+            "Migration " <> Text.pack (show i) <> "/" <> Text.pack (show (length scripts'))
+          applyMigration' Nothing stdout script
+
       putStrLn "Success!"
       pure ranAll
+
     (Just logfiledir, scripts) -> do
-      -- Remove the pattern match here too
       logFilename <- genLogFilename logfiledir
       withFile logFilename AppendMode $ \logHandle -> do
         unless quiet $ putStrLn "Running:"
-        (scripts', ranAll) <- filterMigrations scripts -- Filter ALL scripts including first
-        forM_ scripts' $ applyMigration' (Just logFilename) logHandle
+        (scripts', ranAll) <- filterMigrations scripts
+
+        -- Replace just this forM_ with progress bar
+        withProgress (length scripts') "Database migrations" $ \progressRef -> do
+          forM_ (zip [1 :: Integer ..] scripts') $ \(i, script) -> do
+            updateProgress progressRef (fromIntegral i) $
+              "Migration " <> Text.pack (show i) <> "/" <> Text.pack (show (length scripts'))
+            applyMigration' (Just logFilename) logHandle script
+
         unless quiet $ putStrLn "Success!"
         pure ranAll
   pure (ranAll, map (takeFileName . snd) (filter isUnofficialMigration allScripts))
@@ -169,37 +179,32 @@ validateMigrations migrationDir knownMigrations = do
     stage3or4 = flip elem [3, 4] . readStageFromFilename . Text.unpack . mvFilepath
 
 applyMigration :: MigrationDir -> Bool -> PGConfig -> Maybe FilePath -> Handle -> (MigrationVersion, FilePath) -> IO ()
-applyMigration (MigrationDir location) quiet pgconfig mLogFilename logHandle (version, script) = do
-  -- This assumes that the credentials for 'psql' are already sorted out.
-  -- One way to achive this is via a 'PGPASSFILE' environment variable
-  -- as per the PostgreSQL documentation.
-  let command =
-        List.unwords
-          [ "psql"
-          , Text.unpack (pgcDbname pgconfig)
-          , "--no-password"
-          , "--quiet"
-          , "--username=" <> Text.unpack (pgcUser pgconfig)
-          , "--host=" <> Text.unpack (pgcHost pgconfig)
-          , "--port=" <> Text.unpack (pgcPort pgconfig)
-          , "--no-psqlrc" -- Ignore the ~/.psqlrc file.
-          , "--single-transaction" -- Run the file as a transaction.
-          , "--set ON_ERROR_STOP=on" -- Exit with non-zero on error.
-          , "--file='" ++ location </> script ++ "'"
-          , "2>&1" -- Pipe stderr to stdout.
-          ]
+applyMigration (MigrationDir location) quiet pgconfig mLogFilename logHandle (_, script) = do
   hPutStrLn logHandle $ "Running : " ++ script
   unless quiet $ putStr ("    " ++ script ++ " ... ")
-  hFlush stdout
-  exitCode <-
-    fst
-      <$> handle
-        (errorExit :: SomeException -> IO a)
-        (runResourceT $ sourceCmdWithConsumer command (sinkHandle logHandle))
+  -- hFlush stdout
+
+  let psqlArgs = [ Text.unpack (pgcDbname pgconfig)
+                 , "--no-password"
+                 , "--quiet"
+                 , "--username=" <> Text.unpack (pgcUser pgconfig)
+                 , "--host=" <> Text.unpack (pgcHost pgconfig)
+                 , "--port=" <> Text.unpack (pgcPort pgconfig)
+                 , "--no-psqlrc"
+                 , "--single-transaction"
+                 , "--set", "ON_ERROR_STOP=on"
+                 , "--file=" ++ location </> script
+                 ]
+
+  hPutStrLn logHandle $ "DEBUG: About to execute psql with args: " ++ show psqlArgs
+  (exitCode, stdt, stderr) <- readProcessWithExitCode "psql" psqlArgs ""
+  hPutStrLn logHandle $ "DEBUG: Command completed with exit code: " ++ show exitCode
+  hPutStrLn logHandle $ "Command output: " ++ stdt
+  unless (null stderr) $ hPutStrLn logHandle $ "Command stderr: " ++ stderr
+
   case exitCode of
     ExitSuccess -> do
       unless quiet $ putStrLn "ok"
-      runHaskellMigration (PGPassCached pgconfig) logHandle version
     ExitFailure _ -> errorExit exitCode
   where
     errorExit :: Show e => e -> IO a
@@ -212,8 +217,6 @@ applyMigration (MigrationDir location) quiet pgconfig mLogFilename logHandle (ve
       exitFailure
 
 -- | Create a database migration.
--- TODO: Cmdv - This functionality will need to be reimplemented without Persistent.
--- For now, this serves as a placeholder.
 createMigration :: PGPassSource -> MigrationDir -> TxOutVariantType -> IO (Maybe FilePath)
 createMigration _source (MigrationDir _migdir) _txOutVariantType = do
   -- This would need to be completely rewritten to generate migrations manually

cardano-db/src/Cardano/Db/Run.hs

@@ -42,6 +42,8 @@ import Cardano.Db.Error (DbCallStack (..), DbError (..), runOrThrowIO)
 import Cardano.Db.PGConfig
 import Cardano.Db.Statement.Function.Core (mkDbCallStack)
 import Cardano.Db.Types (DbAction (..), DbEnv (..))
+import Cardano.Db.Statement (runDbSession)
+import qualified Hasql.Session as HsqlSess
 
 -----------------------------------------------------------------------------------------
 -- Transaction Management
@@ -73,6 +75,11 @@ commitTransactionStmt :: HsqlStmt.Statement () ()
 commitTransactionStmt =
   HsqlStmt.Statement "COMMIT" HsqlE.noParams HsqlD.noResult True
 
+commitCurrentTransaction :: MonadIO m => DbAction m ()
+commitCurrentTransaction = do
+  runDbSession (mkDbCallStack "commitCurrentTransaction") $
+    HsqlSess.statement () commitTransactionStmt
+
 -- | Rollback transaction
 rollbackTransactionStmt :: HsqlStmt.Statement () ()
 rollbackTransactionStmt =