fix timestamp conversion and rebase errors

Author: Cmdv

cabal.project

@@ -10,13 +10,8 @@ repository cardano-haskell-packages
     d4a35cd3121aa00d18544bb0ac01c3e1691d618f462c46129271bccf39f7e8ee
 
 index-state:
-<<<<<<< HEAD
   , hackage.haskell.org 2025-05-23T06:30:40Z
   , cardano-haskell-packages 2025-05-16T20:03:45Z
-=======
-  , hackage.haskell.org 2025-02-05T12:01:20Z
-  , cardano-haskell-packages 2025-02-04T11:56:25Z
->>>>>>> 38627e31 (make a start to stetting up hasql with db pools)
 
 packages:
   cardano-db

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

@@ -31,11 +31,12 @@ import qualified Data.Text as Text
 import Data.Version (showVersion)
 import qualified Hasql.Connection as HsqlC
 import qualified Hasql.Connection.Setting as HsqlSet
-import qualified Ouroboros.Consensus.HardFork.Simple as HardFork
+import Ouroboros.Consensus.Cardano (CardanoHardForkTrigger (..))
 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
@@ -58,7 +59,6 @@ import Cardano.DbSync.Tracing.ToObjectOrphans ()
 import Cardano.DbSync.Types
 import Cardano.Prelude hiding (Nat, (%))
 import Cardano.Slotting.Slot (EpochNo (..))
-import Control.Concurrent.Async
 
 runDbSyncNode :: MetricSetters -> [(Text, Text)] -> SyncNodeParams -> SyncNodeConfig -> IO ()
 runDbSyncNode metricsSetters knownMigrations params syncNodeConfigFromFile =
@@ -240,15 +240,14 @@ runSyncNode metricsSetters trce iomgr dbConnSetting runDelayedMigrationFnc syncN
 
           -- communication channel between datalayer thread and chainsync-client thread
           threadChannels <- liftIO newThreadChannels
-          liftIO $
-            mapConcurrently_
-              id
-              [ runDbThread syncEnv metricsSetters threadChannels
-              , 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)
+import Cardano.Prelude (textShow, MonadError (..))
 import Control.Concurrent.Class.MonadSTM.Strict (atomically, readTVarIO, writeTVar)
 import Control.Monad.Extra
 import Control.Monad.IO.Class (MonadIO, liftIO)
@@ -156,7 +156,10 @@ prepareTxOut ::
 prepareTxOut syncEnv (TxIn txIntxId (TxIx index), txOut) = do
   let txHashByteString = Generic.safeHashToByteString $ unTxId txIntxId
   let genTxOut = fromTxOut (fromIntegral index) txOut
-  txId <- queryTxIdWithCache cache txIntxId
+  eTxId <- queryTxIdWithCache cache txIntxId
+  txId <- case eTxId of
+    Left err -> throwError err
+    Right tid -> pure tid
   insertTxOut trce cache iopts (txId, txHashByteString) genTxOut
   where
     trce = getTrace syncEnv

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

@@ -407,15 +407,46 @@ queryMAWithCache cache policyId asset =
       let !assetNameBs = Generic.unAssetName asset
       maybe (Left (policyBs, assetNameBs)) Right <$> DB.queryMultiAssetId policyBs assetNameBs
 
+-- CORRECT VERSION - match the original cache behavior exactly:
+
 queryTxIdWithCacheEither ::
   MonadIO m =>
   CacheStatus ->
-  Ledger.TxId StandardCrypto ->
+  Ledger.TxId ->  -- Use the original input type
   DB.DbAction m (Either DB.DbError DB.TxId)
 queryTxIdWithCacheEither cache txIdLedger = do
-  catchError
-    (Right <$> queryTxIdWithCache cache txIdLedger)
-    (pure . Left)
+  case cache of
+    -- Direct database query if no cache.
+    NoCache -> qTxHash
+    ActiveCache ci ->
+      withCacheOptimisationCheck ci qTxHash $ do
+        -- Read current cache state.
+        cacheTx <- liftIO $ readTVarIO (cTxIds ci)
+
+        case FIFO.lookup txIdLedger cacheTx of
+          -- Cache hit, return the transaction ID.
+          Just txId -> do
+            liftIO $ hitTxIds (cStats ci)
+            pure $ Right txId
+          -- Cache miss.
+          Nothing -> do
+            eTxId <- qTxHash
+            liftIO $ missTxIds (cStats ci)
+            case eTxId of
+              Right txId -> do
+                -- Update cache.
+                liftIO $ atomically $ modifyTVar (cTxIds ci) $ FIFO.insert txIdLedger txId
+                -- Return ID after updating cache.
+                pure $ Right txId
+              -- Return lookup failure.
+              Left err -> pure $ Left err
+  where
+    txHash = Generic.unTxHash txIdLedger  -- Convert to ByteString for DB query
+    qTxHash = do
+      result <- DB.queryTxId txHash
+      case result of
+        Just txId -> pure $ Right txId
+        Nothing -> pure $ Left $ DB.DbError (DB.mkDbCallStack "queryTxIdWithCacheEither") "TxId not found" Nothing
 
 queryPrevBlockWithCache ::
   MonadIO m =>
@@ -450,7 +481,7 @@ queryTxIdWithCache ::
   MonadIO m =>
   CacheStatus ->
   Ledger.TxId ->
-  DB.DbAction m DB.TxId
+  DB.DbAction m (Either DB.DbError DB.TxId)
 queryTxIdWithCache cache txIdLedger = do
   case cache of
     -- Direct database query if no cache.
@@ -459,22 +490,32 @@ queryTxIdWithCache cache txIdLedger = do
       withCacheOptimisationCheck ci qTxHash $ do
         -- Read current cache state.
         cacheTx <- liftIO $ readTVarIO (cTxIds ci)
+
         case FIFO.lookup txIdLedger cacheTx of
           -- Cache hit, return the transaction ID.
           Just txId -> do
             liftIO $ hitTxIds (cStats ci)
-            pure txId
+            pure $ Right txId
           -- Cache miss.
           Nothing -> do
-            txId <- qTxHash
+            eTxId <- qTxHash
             liftIO $ missTxIds (cStats ci)
-            -- Update cache.
-            liftIO $ atomically $ modifyTVar (cTxIds ci) $ FIFO.insert txIdLedger txId
-            -- Return ID after updating cache.
-            pure txId
+            case eTxId of
+              Right txId -> do
+                -- Update cache ONLY on successful lookup.
+                liftIO $ atomically $ modifyTVar (cTxIds ci) $ FIFO.insert txIdLedger txId
+                -- Return ID after updating cache.
+                pure $ Right txId
+              -- Return lookup failure - DON'T update cache.
+              Left err -> pure $ Left err
   where
     txHash = Generic.unTxHash txIdLedger
-    qTxHash = DB.queryTxId txHash
+    qTxHash = 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
 
 tryUpdateCacheTx ::
   MonadIO m =>

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

@@ -4,21 +4,16 @@
 {-# LANGUAGE NoImplicitPrelude #-}
 
 module Cardano.DbSync.Database (
-  DbEvent (..),
-  ThreadChannels,
-  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.Api.Types (ConsistentLevel (..), SyncEnv (..))
 import Cardano.DbSync.DbEvent
 import Cardano.DbSync.Default
 import Cardano.DbSync.Error
 import Cardano.DbSync.Ledger.State
-import Cardano.DbSync.Ledger.Types (CardanoLedgerState (..), SnapshotPoint (..))
 import Cardano.DbSync.Metrics
 import Cardano.DbSync.Rollback
 import Cardano.DbSync.Types
@@ -27,9 +22,6 @@ import Cardano.Prelude hiding (atomically)
 import Cardano.Slotting.Slot (WithOrigin (..))
 import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Monad.Extra (whenJust)
-import Control.Monad.Trans.Except.Extra (newExceptT)
-import Ouroboros.Consensus.HeaderValidation hiding (TipInfo)
-import Ouroboros.Consensus.Ledger.Extended
 import Ouroboros.Network.Block (BlockNo, Point (..))
 import Ouroboros.Network.Point (blockPointHash, blockPointSlot)
 
@@ -75,7 +67,7 @@ runDbThread syncEnv metricsSetters queue = do
 
       -- Handle the result of running the actions
       case result of
-        Left err -> logError tracer $ "Error: " <> show err
+        Left err -> logError tracer $ show err
         Right Continue -> processQueue -- Continue processing
         Right Done -> pure () -- Stop processing
 
@@ -114,55 +106,31 @@ runActions syncEnv actions = do
         ([], DbFinish : _) -> do
           pure Done
         ([], DbRollBackToPoint chainSyncPoint serverTip resultVar : ys) -> do
-          deletedAllBlocks <- prepareRollback syncEnv chainSyncPoint serverTip
+          -- Fix: prepareRollback now returns IO (Either SyncNodeError Bool), so use ExceptT
+          deletedAllBlocks <- ExceptT $ prepareRollback syncEnv chainSyncPoint serverTip
           points <- lift $ rollbackLedger syncEnv chainSyncPoint
 
-          -- Ledger state always rollbacks at least back to the 'point' given by the Node.
-          -- It needs to rollback even further, if 'points' is not 'Nothing'.
-          -- The db may not rollback to the Node point.
+          -- Keep the same logic as before for consistency levels
           case (deletedAllBlocks, points) of
             (True, Nothing) -> do
               liftIO $ setConsistentLevel syncEnv Consistent
               liftIO $ validateConsistentLevel syncEnv chainSyncPoint
             (False, Nothing) -> do
               liftIO $ setConsistentLevel syncEnv DBAheadOfLedger
               liftIO $ validateConsistentLevel syncEnv chainSyncPoint
-            _anyOtherOption ->
+            _anyOtherOption -> do
               -- No need to validate here
               liftIO $ setConsistentLevel syncEnv DBAheadOfLedger
           blockNo <- lift $ getDbTipBlockNo syncEnv
           lift $ atomically $ putTMVar resultVar (points, blockNo)
           dbEvent Continue ys
         (ys, zs) -> do
-          newExceptT $ insertListBlocks syncEnv ys
+          -- Fix: insertListBlocks now returns IO (Either SyncNodeError ()), so use ExceptT
+          ExceptT $ insertListBlocks syncEnv ys
           if null zs
             then pure Continue
             else dbEvent Continue zs
 
-rollbackLedger :: SyncEnv -> CardanoPoint -> IO (Maybe [CardanoPoint])
-rollbackLedger syncEnv point =
-  case envLedgerEnv syncEnv of
-    HasLedger hle -> do
-      mst <- loadLedgerAtPoint hle point
-      case mst of
-        Right st -> do
-          let statePoint = headerStatePoint $ headerState $ clsState st
-          -- This is an extra validation that should always succeed.
-          unless (point == statePoint) $
-            logAndThrowIO (getTrace syncEnv) $
-              SNErrDatabaseRollBackLedger $
-                mconcat
-                  [ "Ledger "
-                  , show statePoint
-                  , " and ChainSync "
-                  , show point
-                  , " don't match."
-                  ]
-          pure Nothing
-        Left lsfs ->
-          Just . fmap fst <$> verifySnapshotPoint syncEnv (OnDisk <$> lsfs)
-    NoLedger _ -> pure Nothing
-
 -- | This not only checks that the ledger and ChainSync points are equal, but also that the
 -- 'Consistent' Level is correct based on the db tip.
 validateConsistentLevel :: SyncEnv -> CardanoPoint -> IO ()