Improve concurrency and exceptions handling

Fixes #1438.
When connection with node restarts, the db-sync threads are no
longer killed. Also when an exception is thrown in a DBSync thread,
it is no longer caught by the chainsync client

Author: kderme

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

@@ -124,6 +124,7 @@ library
 
   build-depends:        base                            >= 4.14         && < 4.17
                       , aeson
+                      , async
                       , binary
                       , bytestring
                       , base16-bytestring

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

@@ -27,20 +27,26 @@ module Cardano.DbSync (
 ) where
 
 import Cardano.BM.Trace (Trace, logError, logInfo, logWarning)
+import qualified Cardano.Crypto as Crypto
 import Cardano.Db (textShow)
 import qualified Cardano.Db as Db
 import Cardano.DbSync.Api
-import Cardano.DbSync.Api.Types (InsertOptions (..), SyncOptions (..))
-import Cardano.DbSync.Config (configureLogging)
+import Cardano.DbSync.Api.Types (InsertOptions (..), RunMigration, SyncOptions (..), envLedgerEnv)
+import Cardano.DbSync.Config (configureLogging, readSyncNodeConfig)
+import Cardano.DbSync.Config.Cardano
 import Cardano.DbSync.Config.Types (
   ConfigFile (..),
   GenesisFile (..),
   LedgerStateDir (..),
   NetworkName (..),
   SocketPath (..),
   SyncCommand (..),
+  SyncNodeConfig (..),
   SyncNodeParams (..),
  )
+import Cardano.DbSync.Database
+import Cardano.DbSync.DbAction
+import Cardano.DbSync.Era
 import Cardano.DbSync.Era.Shelley.Offline.Http (
   FetchError (..),
   SimplifiedPoolOfflineData (..),
@@ -49,19 +55,27 @@ import Cardano.DbSync.Era.Shelley.Offline.Http (
   renderFetchError,
   spodJson,
  )
+import Cardano.DbSync.Error
+import Cardano.DbSync.Ledger.State
 import Cardano.DbSync.Rollback (unsafeRollback)
-import Cardano.DbSync.Sync (runSyncNode)
+import Cardano.DbSync.Sync (runSyncNodeClient)
 import Cardano.DbSync.Tracing.ToObjectOrphans ()
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util (readAbortOnPanic)
 import Cardano.Prelude hiding (Nat, (%))
+import Cardano.Slotting.Slot (EpochNo (..))
+import Control.Concurrent.Async
 import Control.Monad.Extra (whenJust)
 import Control.Monad.Trans.Except.Exit (orDie)
 import Control.Monad.Trans.Except.Extra (newExceptT)
 import qualified Data.Text as Text
 import Data.Version (showVersion)
+import Database.Persist.Postgresql (ConnectionString, withPostgresqlConn)
+import qualified Ouroboros.Consensus.HardFork.Simple as HardFork
 import Ouroboros.Network.NodeToClient (IOManager, withIOManager)
 import Paths_cardano_db_sync (version)
+import System.Directory (createDirectoryIfMissing)
+import Prelude (id)
 
 runDbSyncNode :: MetricSetters -> [(Text, Text)] -> SyncNodeParams -> IO ()
 runDbSyncNode metricsSetters knownMigrations params =
@@ -137,6 +151,78 @@ runDbSync metricsSetters knownMigrations iomgr trce params aop = do
 
     syncOpts = extractSyncOptions params aop
 
+runSyncNode ::
+  MetricSetters ->
+  Trace IO Text ->
+  IOManager ->
+  ConnectionString ->
+  -- | migrations were ran on startup
+  Bool ->
+  -- | run migration function
+  RunMigration ->
+  SyncNodeParams ->
+  SyncOptions ->
+  IO ()
+runSyncNode metricsSetters trce iomgr dbConnString ranMigrations runMigrationFnc syncNodeParams syncOptions = do
+  syncNodeConfig <- readSyncNodeConfig configFile
+  whenJust maybeLedgerDir $
+    \enpLedgerStateDir -> do
+      createDirectoryIfMissing True (unLedgerStateDir enpLedgerStateDir)
+
+  logInfo trce $ "Using byron genesis file from: " <> (show . unGenesisFile $ dncByronGenesisFile syncNodeConfig)
+  logInfo trce $ "Using shelley genesis file from: " <> (show . unGenesisFile $ dncShelleyGenesisFile syncNodeConfig)
+  logInfo trce $ "Using alonzo genesis file from: " <> (show . unGenesisFile $ dncAlonzoGenesisFile syncNodeConfig)
+  Db.runIohkLogging trce $
+    withPostgresqlConn dbConnString $ \backend -> liftIO $ do
+      orDie renderSyncNodeError $ do
+        genCfg <- readCardanoGenesisConfig syncNodeConfig
+        logProtocolMagicId trce $ genesisProtocolMagicId genCfg
+
+        syncEnv <-
+          ExceptT $
+            mkSyncEnvFromConfig
+              trce
+              dbConnString
+              backend
+              syncOptions
+              genCfg
+              syncNodeParams
+              ranMigrations
+              runMigrationFnc
+        liftIO $ runExtraMigrationsMaybe syncEnv
+        unless (enpShouldUseLedger syncNodeParams) $ liftIO $ do
+          logInfo trce "Migrating to a no ledger schema"
+          Db.noLedgerMigrations backend trce
+        insertValidateGenesisDist syncEnv (dncNetworkName syncNodeConfig) genCfg (useShelleyInit syncNodeConfig)
+
+        -- 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)
+            , runOfflineFetchThread syncEnv
+            , runLedgerStateWriteThread (getTrace syncEnv) (envLedgerEnv syncEnv)
+            ]
+  where
+    useShelleyInit :: SyncNodeConfig -> Bool
+    useShelleyInit cfg =
+      case dncShelleyHardFork cfg of
+        HardFork.TriggerHardForkAtEpoch (EpochNo 0) -> True
+        _ -> False
+
+    configFile = enpConfigFile syncNodeParams
+    maybeLedgerDir = enpMaybeLedgerStateDir syncNodeParams
+
+logProtocolMagicId :: Trace IO Text -> Crypto.ProtocolMagicId -> ExceptT SyncNodeError IO ()
+logProtocolMagicId tracer pm =
+  liftIO . logInfo tracer $
+    mconcat
+      [ "NetworkMagic: "
+      , textShow (Crypto.unProtocolMagicId pm)
+      ]
+
 -- -------------------------------------------------------------------------------------------------
 
 extractSyncOptions :: SyncNodeParams -> Bool -> SyncOptions

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

@@ -29,9 +29,7 @@ module Cardano.DbSync.Api (
   getHasConsumed,
   getPrunes,
   mkSyncEnvFromConfig,
-  replaceConnection,
   verifySnapshotPoint,
-  getBackend,
   getInsertOptions,
   getTrace,
   getTopLevelConfig,
@@ -155,8 +153,7 @@ runExtraMigrationsMaybe env = do
   extraMigr <- liftIO $ readTVarIO $ envExtraMigrations env
   logInfo (getTrace env) $ textShow extraMigr
   unless (emRan extraMigr) $ do
-    backend <- getBackend env
-    DB.runDbIohkNoLogging backend $
+    DB.runDbIohkNoLogging (envBackend env) $
       DB.runExtraMigrations
         (getTrace env)
         (getSafeBlockNoDiff env)
@@ -199,10 +196,6 @@ defaultInsertOptions = fullInsertOptions
 turboInsertOptions :: InsertOptions
 turboInsertOptions = InsertOptions False False False False
 
-replaceConnection :: SyncEnv -> SqlBackend -> IO ()
-replaceConnection env sqlBackend = do
-  atomically $ writeTVar (envBackend env) $ Strict.Just sqlBackend
-
 initEpochState :: EpochState
 initEpochState =
   EpochState
@@ -259,13 +252,6 @@ getInsertOptions = soptInsertOptions . envOptions
 getSlotHash :: SqlBackend -> SlotNo -> IO [(SlotNo, ByteString)]
 getSlotHash backend = DB.runDbIohkNoLogging backend . DB.querySlotHash
 
-getBackend :: SyncEnv -> IO SqlBackend
-getBackend env = do
-  mBackend <- readTVarIO $ envBackend env
-  case mBackend of
-    Strict.Just conn -> pure conn
-    Strict.Nothing -> panic "sql connection not initiated"
-
 hasLedgerState :: SyncEnv -> Bool
 hasLedgerState syncEnv =
   case envLedgerEnv syncEnv of
@@ -287,18 +273,16 @@ getDbLatestBlockInfo backend = do
         }
 
 getDbTipBlockNo :: SyncEnv -> IO (Point.WithOrigin BlockNo)
-getDbTipBlockNo env =
-  getBackend env
-    >>= getDbLatestBlockInfo
-    <&> maybe Point.Origin (Point.At . bBlockNo)
+getDbTipBlockNo env = do
+  mblk <- getDbLatestBlockInfo (envBackend env)
+  pure $ maybe Point.Origin (Point.At . bBlockNo) mblk
 
 logDbState :: SyncEnv -> IO ()
 logDbState env = do
-  backend <- getBackend env
-  mblk <- getDbLatestBlockInfo backend
+  mblk <- getDbLatestBlockInfo (envBackend env)
   case mblk of
-    Nothing -> logInfo tracer "Cardano.Db is empty"
-    Just tip -> logInfo tracer $ mconcat ["Cardano.Db tip is at ", showTip tip]
+    Nothing -> logInfo tracer "Database is empty"
+    Just tip -> logInfo tracer $ mconcat ["Database tip is at ", showTip tip]
   where
     showTip :: TipInfo -> Text
     showTip tipInfo =
@@ -314,15 +298,15 @@ logDbState env = do
 
 getCurrentTipBlockNo :: SyncEnv -> IO (WithOrigin BlockNo)
 getCurrentTipBlockNo env = do
-  backend <- getBackend env
-  maybeTip <- getDbLatestBlockInfo backend
+  maybeTip <- getDbLatestBlockInfo (envBackend env)
   case maybeTip of
     Just tip -> pure $ At (bBlockNo tip)
     Nothing -> pure Origin
 
 mkSyncEnv ::
   Trace IO Text ->
   ConnectionString ->
+  SqlBackend ->
   SyncOptions ->
   ProtocolInfo IO CardanoBlock ->
   Ledger.Network ->
@@ -332,9 +316,8 @@ mkSyncEnv ::
   Bool ->
   RunMigration ->
   IO SyncEnv
-mkSyncEnv trce connString syncOptions protoInfo nw nwMagic systemStart syncNodeParams ranMigrations runMigrationFnc = do
+mkSyncEnv trce connString backend syncOptions protoInfo nw nwMagic systemStart syncNodeParams ranMigrations runMigrationFnc = do
   cache <- if soptCache syncOptions then newEmptyCache 250000 50000 else pure uninitiatedCache
-  backendVar <- newTVarIO Strict.Nothing
   consistentLevelVar <- newTVarIO Unchecked
   fixDataVar <- newTVarIO $ if ranMigrations then DataFixRan else NoneFixRan
   indexesVar <- newTVarIO $ enpForceIndexes syncNodeParams
@@ -370,7 +353,7 @@ mkSyncEnv trce connString syncOptions protoInfo nw nwMagic systemStart syncNodeP
       , envSystemStart = systemStart
       , envConnString = connString
       , envRunDelayedMigration = runMigrationFnc
-      , envBackend = backendVar
+      , envBackend = backend
       , envOptions = syncOptions
       , envConsistentLevel = consistentLevelVar
       , envIsFixed = fixDataVar
@@ -387,6 +370,7 @@ mkSyncEnv trce connString syncOptions protoInfo nw nwMagic systemStart syncNodeP
 mkSyncEnvFromConfig ::
   Trace IO Text ->
   ConnectionString ->
+  SqlBackend ->
   SyncOptions ->
   GenesisConfig ->
   SyncNodeParams ->
@@ -395,7 +379,7 @@ mkSyncEnvFromConfig ::
   -- | run migration function
   RunMigration ->
   IO (Either SyncNodeError SyncEnv)
-mkSyncEnvFromConfig trce connString syncOptions genCfg syncNodeParams ranMigration runMigrationFnc =
+mkSyncEnvFromConfig trce connString backend syncOptions genCfg syncNodeParams ranMigration runMigrationFnc =
   case genCfg of
     GenesisCardano _ bCfg sCfg _
       | unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -419,6 +403,7 @@ mkSyncEnvFromConfig trce connString syncOptions genCfg syncNodeParams ranMigrati
             <$> mkSyncEnv
               trce
               connString
+              backend
               syncOptions
               (mkProtocolInfoCardano genCfg [])
               (Shelley.sgNetworkId $ scConfig sCfg)
@@ -437,8 +422,7 @@ getLatestPoints env = do
       verifySnapshotPoint env snapshotPoints
     NoLedger _ -> do
       -- Brings the 5 latest.
-      dbBackend <- getBackend env
-      lastPoints <- DB.runDbIohkNoLogging dbBackend DB.queryLatestPoints
+      lastPoints <- DB.runDbIohkNoLogging (envBackend env) DB.queryLatestPoints
       pure $ mapMaybe convert lastPoints
   where
     convert (Nothing, _) = Nothing
@@ -450,8 +434,7 @@ verifySnapshotPoint env snapPoints =
   where
     validLedgerFileToPoint :: SnapshotPoint -> IO (Maybe (CardanoPoint, Bool))
     validLedgerFileToPoint (OnDisk lsf) = do
-      backend <- getBackend env
-      hashes <- getSlotHash backend (lsfSlotNo lsf)
+      hashes <- getSlotHash (envBackend env) (lsfSlotNo lsf)
       let valid = find (\(_, h) -> lsfHash lsf == hashToAnnotation h) hashes
       case valid of
         Just (slot, hash) | slot == lsfSlotNo lsf -> pure $ convertToDiskPoint slot hash
@@ -460,8 +443,7 @@ verifySnapshotPoint env snapPoints =
       case pnt of
         GenesisPoint -> pure Nothing
         BlockPoint slotNo hsh -> do
-          backend <- getBackend env
-          hashes <- getSlotHash backend slotNo
+          hashes <- getSlotHash (envBackend env) slotNo
           let valid = find (\(_, dbHash) -> getHeaderHash hsh == dbHash) hashes
           case valid of
             Just (dbSlotNo, _) | slotNo == dbSlotNo -> pure $ Just (pnt, True)

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

@@ -39,7 +39,7 @@ data SyncEnv = SyncEnv
   , envSystemStart :: !SystemStart
   , envConnString :: ConnectionString
   , envRunDelayedMigration :: RunMigration
-  , envBackend :: !(StrictTVar IO (Strict.Maybe SqlBackend))
+  , envBackend :: !SqlBackend
   , envConsistentLevel :: !(StrictTVar IO ConsistentLevel)
   , envIsFixed :: !(StrictTVar IO FixesRan)
   , envIndexes :: !(StrictTVar IO Bool)