start working way through test to check conversion correct

Author: Cmdv

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

@@ -207,7 +207,7 @@ startDBSync env = do
     Just _a -> error "db-sync already running"
     Nothing -> do
       let appliedRunDbSync = partialRunDbSync env (dbSyncParams env) (dbSyncConfig env)
-      -- we async the fully applied runDbSync here ad put it into the thread
+      -- we async the fully applied runDbSync here and put it into the thread
       asyncApplied <- async appliedRunDbSync
       void . atomically $ tryPutTMVar (dbSyncThreadVar env) asyncApplied
 
@@ -225,7 +225,8 @@ getDBSyncPGPass :: DBSyncEnv -> DB.PGPassSource
 getDBSyncPGPass = enpPGPassSource . dbSyncParams
 
 queryDBSync :: DBSyncEnv -> DB.DbAction (NoLoggingT IO) a -> IO a
-queryDBSync env = DB.runWithConnectionNoLogging (getDBSyncPGPass env)
+queryDBSync env = do
+  DB.runWithConnectionNoLogging (getDBSyncPGPass env)
 
 getPoolLayer :: DBSyncEnv -> IO PoolDataLayer
 getPoolLayer env = do
@@ -557,7 +558,7 @@ withFullConfig' WithConfigArgs {..} cmdLineArgs mSyncNodeConfig configFilePath t
       then configureLogging syncNodeConfig "db-sync-node"
       else pure nullTracer
   -- runDbSync is partially applied so we can pass in syncNodeParams at call site / within tests
-  let partialDbSyncRun params cfg' = runDbSync emptyMetricsSetters migr iom trce params cfg' True
+  let partialDbSyncRun params cfg' = runDbSync emptyMetricsSetters iom trce params cfg' True
       initSt = Consensus.pInfoInitLedger $ protocolInfo cfg
 
   withInterpreter (protocolInfoForging cfg) (protocolInfoForger cfg) nullTracer fingerFile $ \interpreter -> do
@@ -578,6 +579,14 @@ withFullConfig' WithConfigArgs {..} cmdLineArgs mSyncNodeConfig configFilePath t
           if null tableNames || shouldDropDB
             then void . hSilence [stderr] $ DB.recreateDB pgPass
             else void . hSilence [stderr] $ DB.truncateTables pgPass tableNames
+
+          -- Run migrations synchronously first
+          runMigrationsOnly
+            migr
+            trce
+            (syncNodeParams cfg)
+            syncNodeConfig
+
           action interpreter mockServer dbSyncEnv
   where
     mutableDir = mkMutableDir testLabelFilePath

cardano-chain-gen/test/Test/Cardano/Db/Mock/Unit/Conway/Config/JsonbInSchema.hs

@@ -20,7 +20,6 @@ configRemoveJsonbFromSchemaEnabled :: IOManager -> [(Text, Text)] -> Assertion
 configRemoveJsonbFromSchemaEnabled = do
   withCustomConfigAndDropDB args (Just configRemoveJsonFromSchema) cfgDir testLabel $ \_interpreter _mockServer dbSync -> do
     startDBSync dbSync
-    threadDelay 7_000_000
     assertEqQuery
       dbSync
       DB.queryJsonbInSchemaExistsTest
@@ -38,7 +37,6 @@ configRemoveJsonbFromSchemaDisabled = do
   withCustomConfigAndDropDB args (Just configRemoveJsonFromSchemaFalse) cfgDir testLabel $
     \_interpreter _mockServer dbSync -> do
       startDBSync dbSync
-      threadDelay 7_000_000
       assertEqQuery
         dbSync
         DB.queryJsonbInSchemaExistsTest
@@ -54,7 +52,6 @@ configJsonbInSchemaShouldRemoveThenAdd :: IOManager -> [(Text, Text)] -> Asserti
 configJsonbInSchemaShouldRemoveThenAdd =
   withCustomConfigAndDropDB args (Just configRemoveJsonFromSchema) cfgDir testLabel $ \_interpreter _mockServer dbSyncEnv -> do
     startDBSync dbSyncEnv
-    threadDelay 7_000_000
     assertEqQuery
       dbSyncEnv
       DB.queryJsonbInSchemaExistsTest
@@ -72,7 +69,6 @@ configJsonbInSchemaShouldRemoveThenAdd =
                   }
             }
     startDBSync newDbSyncEnv
-    threadDelay 7_000_000
     assertEqQuery
       dbSyncEnv
       DB.queryJsonbInSchemaExistsTest

cardano-chain-gen/test/Test/Cardano/Db/Mock/Unit/Conway/Config/MigrateConsumedPruneTxOut.hs

@@ -106,6 +106,7 @@ performPruneWithSimpleRollback useTxOutAddress =
     atomically $ addBlock mockServer blk0
 
     startDBSync dbSync
+
     atomically $ addBlock mockServer blk1
 
     -- Create some payment transactions

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

@@ -17,6 +17,7 @@ module Cardano.DbSync (
   SocketPath (..),
   Db.MigrationDir (..),
   runDbSyncNode,
+  runMigrationsOnly,
   runDbSync,
   -- For testing and debugging
   OffChainFetchError (..),
@@ -67,19 +68,19 @@ runDbSyncNode metricsSetters knownMigrations params syncNodeConfigFromFile =
     abortOnPanic <- hasAbortOnPanicEnv
     startupReport trce abortOnPanic params
 
-    runDbSync metricsSetters knownMigrations iomgr trce params syncNodeConfigFromFile abortOnPanic
+    -- Run initial migrations synchronously first
+    runMigrationsOnly knownMigrations trce params syncNodeConfigFromFile
 
-runDbSync ::
-  MetricSetters ->
+    runDbSync metricsSetters iomgr trce params syncNodeConfigFromFile abortOnPanic
+
+-- Extract just the initial migration logic (no indexes)
+runMigrationsOnly ::
   [(Text, Text)] ->
-  IOManager ->
   Trace IO Text ->
   SyncNodeParams ->
   SyncNodeConfig ->
-  -- Should abort on panic
-  Bool ->
   IO ()
-runDbSync metricsSetters knownMigrations iomgr trce params syncNodeConfigFromFile abortOnPanic = do
+runMigrationsOnly knownMigrations trce params syncNodeConfigFromFile = do
   logInfo trce $ textShow syncOpts
 
   -- Read the PG connection info
@@ -97,9 +98,11 @@ runDbSync metricsSetters knownMigrations iomgr trce params syncNodeConfigFromFil
         msg <- Db.getMaintenancePsqlConf pgConfig
         logInfo trce $ "Running database migrations in mode " <> textShow mode
         logInfo trce msg
-        when (mode `elem` [Db.Indexes, Db.Full]) $ logWarning trce indexesMsg
+        -- No index warning here - runMigrationsOnly never runs indexes
         Db.runMigrations pgConfig True dbMigrationDir (Just $ Db.LogFileDir "/tmp") mode (txOutConfigToTableType txOutConfig)
-  (ranMigrations, unofficial) <- if enpForceIndexes params then runMigration Db.Full else runMigration Db.Initial
+
+  -- Always run Initial mode only - never indexes
+  (ranMigrations, unofficial) <- runMigration Db.Initial
   unless (null unofficial) $
     logWarning trce $
       "Unofficial migration scripts found: "
@@ -109,9 +112,27 @@ runDbSync metricsSetters knownMigrations iomgr trce params syncNodeConfigFromFil
     then logInfo trce "All migrations were executed"
     else logInfo trce "Some migrations were not executed. They need to run when syncing has started."
 
-  if enpForceIndexes params
-    then logInfo trce "All user indexes were created"
-    else logInfo trce "New user indexes were not created. They may be created later if necessary."
+  logInfo trce "New user indexes were not created. They may be created later if necessary."
+  where
+    dbMigrationDir :: Db.MigrationDir
+    dbMigrationDir = enpMigrationDir params
+    syncOpts = extractSyncOptions params False syncNodeConfigFromFile
+    txOutConfig = sioTxOut $ dncInsertOptions syncNodeConfigFromFile
+
+runDbSync ::
+  MetricSetters ->
+  IOManager ->
+  Trace IO Text ->
+  SyncNodeParams ->
+  SyncNodeConfig ->
+  -- Should abort on panic
+  Bool ->
+  IO ()
+runDbSync metricsSetters iomgr trce params syncNodeConfigFromFile abortOnPanic = do
+  logInfo trce $ textShow syncOpts
+
+  -- Read the PG connection info
+  pgConfig <- runOrThrowIO (Db.readPGPass $ enpPGPassSource params)
 
   dbConnectionSetting <- case Db.toConnectionSetting pgConfig of
     Left err -> do
@@ -123,20 +144,30 @@ runDbSync metricsSetters knownMigrations iomgr trce params syncNodeConfigFromFil
   -- For testing and debugging.
   whenJust (enpMaybeRollback params) $ \slotNo ->
     void $ unsafeRollback trce (txOutConfigToTableType txOutConfig) pgConfig slotNo
+
+  -- This runMigration is ONLY for delayed migrations during sync (like indexes)
+  let runDelayedMigration mode = do
+        msg <- Db.getMaintenancePsqlConf pgConfig
+        logInfo trce $ "Running database migrations in mode " <> textShow mode
+        logInfo trce msg
+        when (mode `elem` [Db.Indexes, Db.Full]) $ logWarning trce indexesMsg
+        Db.runMigrations pgConfig True dbMigrationDir (Just $ Db.LogFileDir "/tmp") mode (txOutConfigToTableType txOutConfig)
+
   runSyncNode
     metricsSetters
     trce
     iomgr
     dbConnectionSetting
-    (void . runMigration)
+    (void . runDelayedMigration)
     syncNodeConfigFromFile
     params
     syncOpts
   where
     dbMigrationDir :: Db.MigrationDir
     dbMigrationDir = enpMigrationDir params
+    syncOpts = extractSyncOptions params abortOnPanic syncNodeConfigFromFile
+    txOutConfig = sioTxOut $ dncInsertOptions syncNodeConfigFromFile
 
-    indexesMsg :: Text
     indexesMsg =
       mconcat
         [ "Creating Indexes. This may require an extended period of time to perform."
@@ -146,10 +177,6 @@ runDbSync metricsSetters knownMigrations iomgr trce params syncNodeConfigFromFil
         , " in the schema directory and restart it."
         ]
 
-    syncOpts = extractSyncOptions params abortOnPanic syncNodeConfigFromFile
-
-    txOutConfig = sioTxOut $ dncInsertOptions syncNodeConfigFromFile
-
 runSyncNode ::
   MetricSetters ->
   Trace IO Text ->
@@ -162,7 +189,7 @@ runSyncNode ::
   SyncNodeParams ->
   SyncOptions ->
   IO ()
-runSyncNode metricsSetters trce iomgr dbConnSetting runMigrationFnc syncNodeConfigFromFile syncNodeParams syncOptions = do
+runSyncNode metricsSetters trce iomgr dbConnSetting runDelayedMigrationFnc syncNodeConfigFromFile syncNodeParams syncOptions = do
   whenJust maybeLedgerDir $
     \enpLedgerStateDir -> do
       createDirectoryIfMissing True (unLedgerStateDir enpLedgerStateDir)
@@ -190,12 +217,11 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runMigrationFnc syncNodeConf
               mkSyncEnvFromConfig
                 trce
                 dbEnv
-                -- dbConnString
                 syncOptions
                 genCfg
                 syncNodeConfigFromFile
                 syncNodeParams
-                runMigrationFnc
+                runDelayedMigrationFnc
 
           -- 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

@@ -315,7 +315,7 @@ mkSyncEnvFromConfig ::
   -- | run migration function
   RunMigration ->
   IO (Either SyncNodeError SyncEnv)
-mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runMigrationFnc =
+mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNodeParams runDelayedMigrationFnc =
   case genCfg of
     GenesisCardano _ bCfg sCfg _ _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -350,7 +350,7 @@ mkSyncEnvFromConfig trce dbEnv syncOptions genCfg syncNodeConfigFromFile syncNod
               (SystemStart . Byron.gdStartTime $ Byron.configGenesisData bCfg)
               syncNodeConfigFromFile
               syncNodeParams
-              runMigrationFnc
+              runDelayedMigrationFnc
 
 mkSyncEnv ::
   Trace IO Text ->
@@ -364,7 +364,7 @@ mkSyncEnv ::
   SyncNodeParams ->
   RunMigration ->
   IO SyncEnv
-mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runMigrationFnc = do
+mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runDelayedMigrationFnc = do
   dbCNamesVar <- newTVarIO =<< DB.runDbActionIO dbEnv DB.queryRewardAndEpochStakeConstraints
   cache <-
     if soptCache syncOptions
@@ -426,7 +426,7 @@ mkSyncEnv trce dbEnv syncOptions protoInfo nw nwMagic systemStart syncNodeConfig
       , envOffChainVoteResultQueue = oarq
       , envOffChainVoteWorkQueue = oawq
       , envOptions = syncOptions
-      , envRunDelayedMigration = runMigrationFnc
+      , envRunDelayedMigration = runDelayedMigrationFnc
       , envSyncNodeConfig = syncNodeConfigFromFile
       , envSystemStart = systemStart
       }

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

@@ -20,6 +20,7 @@ module Cardano.DbSync.Cache (
   insertAddressUsingCache,
   insertStakeAddress,
   queryStakeAddrWithCache,
+  queryTxIdWithCacheEither,
   queryTxIdWithCache,
   rollbackCache,
   optimiseCaches,
@@ -224,7 +225,7 @@ queryPoolKeyWithCache cache cacheUA hsh =
     NoCache -> do
       mPhId <- DB.queryPoolHashId (Generic.unKeyHashRaw hsh)
       case mPhId of
-        Nothing -> pure $ Left $ DB.DbError DB.mkCallSite "queryPoolKeyWithCache: NoCache queryPoolHashId" Nothing
+        Nothing -> pure $ Left $ DB.DbError (DB.mkDbCallStack "queryPoolKeyWithCache") "NoCache queryPoolHashId" Nothing
         Just phId -> pure $ Right phId
     ActiveCache ci -> do
       mp <- liftIO $ readTVarIO (cPools ci)
@@ -242,7 +243,7 @@ queryPoolKeyWithCache cache cacheUA hsh =
           liftIO $ missPools (cStats ci)
           mPhId <- DB.queryPoolHashId (Generic.unKeyHashRaw hsh)
           case mPhId of
-            Nothing -> throwError $ DB.DbError DB.mkCallSite "queryPoolKeyWithCache: ActiveCache queryPoolHashId" Nothing
+            Nothing -> throwError $ DB.DbError (DB.mkDbCallStack "queryPoolKeyWithCache") "ActiveCache queryPoolHashId" Nothing
             Just phId -> do
               -- missed so we can't evict even with 'EvictAndReturn'
               when (shouldCache cacheUA) $
@@ -407,6 +408,16 @@ queryMAWithCache cache policyId asset =
       let !assetNameBs = Generic.unAssetName asset
       maybe (Left (policyBs, assetNameBs)) Right <$> DB.queryMultiAssetId policyBs assetNameBs
 
+queryTxIdWithCacheEither ::
+  MonadIO m =>
+  CacheStatus ->
+  Ledger.TxId StandardCrypto ->
+  DB.DbAction m (Either DB.DbError DB.TxId)
+queryTxIdWithCacheEither cache txIdLedger = do
+  catchError
+    (Right <$> queryTxIdWithCache cache txIdLedger)
+    (pure . Left)
+
 queryPrevBlockWithCache ::
   MonadIO m =>
   CacheStatus ->

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

@@ -69,7 +69,7 @@ writeEpochBlockDiffToCache ::
   DB.DbAction m ()
 writeEpochBlockDiffToCache cache epCurrent =
   case cache of
-    NoCache -> throwError $ DB.DbError DB.mkCallSite "writeEpochBlockDiffToCache: Cache is NoCache" Nothing
+    NoCache -> throwError $ DB.DbError (DB.mkDbCallStack "writeEpochBlockDiffToCache") "Cache is NoCache" Nothing
     ActiveCache ci -> do
       cE <- liftIO $ readTVarIO (cEpoch ci)
       case (ceMapEpoch cE, ceEpochBlockDiff cE) of
@@ -91,12 +91,12 @@ writeToMapEpochCache syncEnv cache latestEpoch = do
           HasLedger hle -> getSecurityParameter $ leProtocolInfo hle
           NoLedger nle -> getSecurityParameter $ nleProtocolInfo nle
   case cache of
-    NoCache -> throwError $ DB.DbError DB.mkCallSite "writeToMapEpochCache: Cache is NoCache" Nothing
+    NoCache -> throwError $ DB.DbError (DB.mkDbCallStack "writeToMapEpochCache") "Cache is NoCache" Nothing
     ActiveCache ci -> do
       -- get EpochBlockDiff so we can use the BlockId we stored when inserting blocks
       epochInternalCE <- readEpochBlockDiffFromCache cache
       case epochInternalCE of
-        Nothing -> throwError $ DB.DbError DB.mkCallSite "writeToMapEpochCache: No epochInternalEpochCache" Nothing
+        Nothing -> throwError $ DB.DbError (DB.mkDbCallStack "writeToMapEpochCache") "No epochInternalEpochCache" Nothing
         Just ei -> do
           cE <- liftIO $ readTVarIO (cEpoch ci)
           let currentBlockId = ebdBlockId ei

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

@@ -144,7 +144,7 @@ textShowStats (ActiveCache ic) = do
   address <- readTVarIO (cAddress ic)
   pure $
     mconcat
-      [ "\nCache Statistics:"
+      [ "\n----------------------- Cache Statistics: -----------------------"
       , "\n  Caches Optimised: " <> textShow isCacheOptimised
       , textCacheSection "Stake Addresses" (scLruCache stakeHashRaws) (scStableCache stakeHashRaws) (credsHits stats) (credsQueries stats)
       , textMapSection "Pools" pools (poolsHits stats) (poolsQueries stats)

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

@@ -67,7 +67,8 @@ runDbThread syncEnv metricsSetters queue = do
     -- Process a list of actions
     processActions :: [DbEvent] -> IO ()
     processActions actions = do
-      result <- runExceptT $ runActions syncEnv actions -- runActions is where we start inserting information we recieve from the node.
+      -- runActions is where we start inserting information we recieve from the node.
+      result <- runExceptT $ runActions syncEnv actions
 
       -- Update metrics with the latest block information
       updateBlockMetrics
@@ -96,46 +97,6 @@ runDbThread syncEnv metricsSetters queue = do
         setDbBlockHeight metricsSetters $ bBlockNo block
         setDbSlotHeight metricsSetters $ bSlotNo block
 
--- runDbThread ::
---   SyncEnv ->
---   MetricSetters ->
---   ThreadChannels ->
---   IO ()
--- runDbThread syncEnv metricsSetters queue = do
---   logInfo trce "Running DB thread"
---   logException trce "runDBThread: " loop
---   logInfo trce "Shutting down DB thread"
---   where
---     trce = getTrace syncEnv
---     loop = do
---       xs <- blockingFlushDbEventQueue queue
-
---       when (length xs > 1) $ do
---         logDebug trce $ "runDbThread: " <> textShow (length xs) <> " blocks"
-
---       case hasRestart xs of
---         Nothing -> do
---           eNextState <- runExceptT $ runActions syncEnv xs
-
---           mBlock <- getDbLatestBlockInfo (envDbEnv syncEnv)
---           whenJust mBlock $ \block -> do
---             setDbBlockHeight metricsSetters $ bBlockNo block
---             setDbSlotHeight metricsSetters $ bSlotNo block
-
---           case eNextState of
---             Left err -> logError trce $ show err
---             Right Continue -> loop
---             Right Done -> pure ()
---         Just resultVar -> do
---           -- In this case the syncing thread has restarted, so ignore all blocks that are not
---           -- inserted yet.
---           logInfo trce "Chain Sync client thread has restarted"
---           latestPoints <- getLatestPoints syncEnv
---           currentTip <- getCurrentTipBlockNo syncEnv
---           logDbState syncEnv
---           atomically $ putTMVar resultVar (latestPoints, currentTip)
---           loop
-
 -- | 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 ::