PLT-7497: restructure Marconi with a focus on logging

marconi-chain-index/bench/BenchQueries.hs

@@ -153,9 +153,9 @@ runIndexerSyncing trace databaseDir nodeSocketPath indexerTVar = do
         networkId
         C.ChainPointAtGenesis
         nodeSocketPath
+        (MinIndexingDepth 0)
+        (ShouldFailIfResync True)
     )
-    (MinIndexingDepth 0)
-    (ShouldFailIfResync True)
     indexers
 
 tests

marconi-chain-index/src/Marconi/ChainIndex/Indexers.hs

@@ -86,6 +86,7 @@ import Control.Monad.Trans.Except (
   ExceptT,
   runExceptT,
  )
+import Control.Monad.Trans.Reader (ReaderT (runReaderT))
 import Data.Functor (($>))
 import Data.List.NonEmpty (NonEmpty)
 import Data.Map (Map)
@@ -110,9 +111,10 @@ import Marconi.ChainIndex.Indexers.EpochState qualified as EpochState
 import Marconi.ChainIndex.Indexers.MintBurn qualified as MintBurn
 import Marconi.ChainIndex.Indexers.ScriptTx qualified as ScriptTx
 import Marconi.ChainIndex.Indexers.Utxo qualified as Utxo
-import Marconi.ChainIndex.Logging (chainSyncEventStreamLogging)
+import Marconi.ChainIndex.Logging (chainSyncEventStreamLogging, logMInfo)
 import Marconi.ChainIndex.Node.Client.Retry (withNodeConnectRetry)
 import Marconi.ChainIndex.Types (
+  ChainIndexerT (ChainIndexerT),
   IndexingDepth (MaxIndexingDepth, MinIndexingDepth),
   RunIndexerConfig (RunIndexerConfig),
   SecurityParam (SecurityParam),
@@ -642,90 +644,81 @@ mkIndexerStream
   -> IO ()
 mkIndexerStream = mkIndexerStream' (CE.bimSlotNo . blockInMode)
 
+-- TODO: placeholder, see definition of 'withNodeConnectRetry'
+withNodeConnectRetryNEW = undefined
+
 runIndexers
-  :: RunIndexerConfig
-  -> IndexingDepth
-  -> ShouldFailIfResync
-  -> [(Worker, Maybe FilePath)]
-  -> IO ()
-runIndexers
-  ( RunIndexerConfig
-      stdoutTrace
-      retryConfig
-      securityParam
-      networkId
-      cliChainPoint
-      socketPath
-    )
-  indexingDepth
-  (ShouldFailIfResync shouldFailIfResync)
-  indexerList = do
-    withNodeConnectRetry stdoutTrace retryConfig socketPath $ do
-      currentNodeBlockNo <- Utils.toException $ Utils.queryCurrentNodeBlockNo @Void networkId socketPath
-      let indexers = mapMaybe sequenceA indexerList
-      coordinator <- initializeCoordinatorFromIndexers securityParam indexingDepth indexers
-      let indexerDepth =
-            let SecurityParam s = securityParam
-             in case indexingDepth of
-                  MinIndexingDepth d -> SecurityParam $ s - d + 1
-                  MaxIndexingDepth -> SecurityParam 1
-      resumablePoints <-
-        getStartingPointsFromIndexers indexerDepth currentNodeBlockNo indexers coordinator
-      let oldestCommonChainPoint = minimum resumablePoints
-          resumePoint = case cliChainPoint of
-            C.ChainPointAtGenesis -> oldestCommonChainPoint -- User didn't specify a chain point, use oldest common chain point,
-            cliCp -> cliCp -- otherwise use what was provided on CLI.
-      logInfo stdoutTrace $
-        "Resumable points for each indexer:"
-          <> line
-          <> indent 4 (align (list (fmap pretty resumablePoints)))
-
-      -- Possible runtime failure if an indexer with a non-genesis resumable point will resume from
-      -- genesis.
-      if shouldFailIfResync
-        && elem C.ChainPointAtGenesis resumablePoints
-        && any (\case ChainPoint{} -> True; _ -> False) resumablePoints
-        then do
-          logError stdoutTrace $
-            nest
-              4
-              ( "At least one indexer has a non-genesis resumable point, while the oldest common resumable point between indexers is genesis."
-                  <> line
-                  <> "Are you sure you want to restart syncing that indexer from genesis?"
-                  <> line
-                  <> "If so, remove the '--fail-if-resyncing-from-genesis' flag."
-              )
-        else do
-          let stream =
-                mkIndexerStream coordinator
-                  . chainSyncEventStreamLogging stdoutTrace
-                  . updateProcessedBlocksMetric
-              runChainSyncStream = withChainSyncBlockEventStream socketPath networkId [resumePoint] stream
-              whenNoIntersectionFound NoIntersectionFound = do
-                logError stdoutTrace $
-                  "No intersection found when looking for the chain point"
-                    <+> pretty resumePoint
+  :: [(Worker, Maybe FilePath)]
+  -> ChainIndexerT IO ()
+runIndexers indexerList = do
+  withNodeConnectRetryNEW $ do
+    -- TODO: get values of config items from Reader, refactor impure functions to run in ChainIndexerT IO
+    --
+    -- currentNodeBlockNo <- Utils.toException $ Utils.queryCurrentNodeBlockNo @Void networkId socketPath
+    -- let indexers = mapMaybe sequenceA indexerList
+    -- coordinator <- initializeCoordinatorFromIndexers securityParam indexingDepth indexers
+    -- let indexerDepth =
+    --       let SecurityParam s = securityParam
+    --        in case indexingDepth of
+    --             MinIndexingDepth d -> SecurityParam $ s - d + 1
+    --             MaxIndexingDepth -> SecurityParam 1
+    -- resumablePoints <-
+    --   getStartingPointsFromIndexers indexerDepth currentNodeBlockNo indexers coordinator
+    -- let oldestCommonChainPoint = minimum resumablePoints
+    --     resumePoint = case cliChainPoint of
+    --       C.ChainPointAtGenesis -> oldestCommonChainPoint -- User didn't specify a chain point, use oldest common chain point,
+    --       cliCp -> cliCp -- otherwise use what was provided on CLI.
+    logMInfo $
+      "Resumable points for each indexer:"
+        <> line
+        <> indent 4 (align (list (fmap pretty resumablePoints)))
+
+    -- Possible runtime failure if an indexer with a non-genesis resumable point will resume from
+    -- genesis.
+    if shouldFailIfResync
+      && elem C.ChainPointAtGenesis resumablePoints
+      && any (\case ChainPoint{} -> True; _ -> False) resumablePoints
+      then do
+        logError stdoutTrace $
+          nest
+            4
+            ( "At least one indexer has a non-genesis resumable point, while the oldest common resumable point between indexers is genesis."
+                <> line
+                <> "Are you sure you want to restart syncing that indexer from genesis?"
+                <> line
+                <> "If so, remove the '--fail-if-resyncing-from-genesis' flag."
+            )
+      else do
+        let stream =
+              mkIndexerStream coordinator
+                . chainSyncEventStreamLogging stdoutTrace
+                . updateProcessedBlocksMetric
+            runChainSyncStream = withChainSyncBlockEventStream socketPath networkId [resumePoint] stream
+            whenNoIntersectionFound NoIntersectionFound = do
+              logError stdoutTrace $
+                "No intersection found when looking for the chain point"
+                  <+> pretty resumePoint
                     <> "."
-                    <+> "Please check the slot number and the block hash do belong to the chain."
-                signalQSemN (coordinator ^. barrier) (coordinator ^. indexerCount)
-           in finally
-                (runChainSyncStream `catch` whenNoIntersectionFound)
-                (waitForIndexersToFinishProcessingLastEvent coordinator)
-    where
-      waitForIndexersToFinishProcessingLastEvent :: Coordinator' a -> IO ()
-      waitForIndexersToFinishProcessingLastEvent coordinator = do
-        let secondsBeforeTimeout = 180
-        logInfo stdoutTrace $
-          "Stopping indexing. Waiting for indexers to finish their work (timeout after "
-            <> pretty secondsBeforeTimeout
-            <> "s) ..."
-        res <-
-          timeout (secondsBeforeTimeout * 1_000_000) $
-            waitQSemN (coordinator ^. barrier) (coordinator ^. indexerCount)
-        case res of
-          Just _ -> logInfo stdoutTrace "Done!"
-          -- TODO: When it's possible, let's put some useful information in this exception
-          Nothing -> throwIO (Timeout @Void "Timed out.")
+                  <+> "Please check the slot number and the block hash do belong to the chain."
+              signalQSemN (coordinator ^. barrier) (coordinator ^. indexerCount)
+         in finally
+              (runChainSyncStream `catch` whenNoIntersectionFound)
+              (waitForIndexersToFinishProcessingLastEvent coordinator)
+  where
+    waitForIndexersToFinishProcessingLastEvent :: Coordinator' a -> IO ()
+    waitForIndexersToFinishProcessingLastEvent coordinator = do
+      let secondsBeforeTimeout = 180
+      logInfo stdoutTrace $
+        "Stopping indexing. Waiting for indexers to finish their work (timeout after "
+          <> pretty secondsBeforeTimeout
+          <> "s) ..."
+      res <-
+        timeout (secondsBeforeTimeout * 1_000_000) $
+          waitQSemN (coordinator ^. barrier) (coordinator ^. indexerCount)
+      case res of
+        Just _ -> logInfo stdoutTrace "Done!"
+        -- TODO: When it's possible, let's put some useful information in this exception
+        Nothing -> throwIO (Timeout @Void "Timed out.")
 
 updateProcessedBlocksMetric
   :: S.Stream (S.Of (ChainSyncEvent BlockEvent)) IO r

marconi-chain-index/src/Marconi/ChainIndex/Logging.hs

@@ -10,6 +10,7 @@ module Marconi.ChainIndex.Logging (
   chainSyncEventStreamLogging,
   MarconiTrace,
   mkMarconiTrace,
+  logMInfo,
 
   -- * Exported for testing purposes
   marconiFormatting,
@@ -22,7 +23,11 @@ import Cardano.Api.Extended.Streaming (
  )
 import Cardano.BM.Trace (Trace, logInfo)
 import Cardano.BM.Tracing (contramap)
+import Control.Lens.Getter qualified as Lens
 import Control.Monad (when)
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Reader (asks)
+import Control.Monad.Trans (lift)
 import Data.IORef (IORef, modifyIORef', newIORef, readIORef)
 import Data.Text (Text)
 import Data.Time (
@@ -36,7 +41,7 @@ import Data.Time (
 import Data.Word (Word64)
 import GHC.Generics (Generic)
 import Marconi.ChainIndex.Orphans ()
-import Marconi.ChainIndex.Types (MarconiTrace)
+import Marconi.ChainIndex.Types (ChainIndexerT, MarconiTrace, runIndexerConfigTrace)
 import Prettyprinter (Pretty (pretty), (<+>))
 import Prettyprinter qualified as Pretty
 import Prettyprinter.Render.Text qualified as Pretty
@@ -54,6 +59,14 @@ marconiFormatting =
   Pretty.renderStrict
     . Pretty.layoutPretty Pretty.defaultLayoutOptions
 
+logMInfo :: MonadIO m => Pretty.Doc () -> ChainIndexerT m ()
+logMInfo msg = do
+  trace <- Lens.view runIndexerConfigTrace
+  lift $ logInfo trace msg
+
+-- TODO: this should also throw an exception
+logMError = undefined
+
 -- | Chain synchronisation statistics measured starting from previously measured 'LastSyncStats'.
 data LastSyncStats = LastSyncStats
   { syncStatsNumBlocks :: !Word64
@@ -87,7 +100,7 @@ instance Pretty LastSyncLog where
               <+> pretty cp
               <+> "and current node tip is"
               <+> pretty nt
-              <> "."
+                <> "."
 
           processingSummaryMsg timeSinceLastMsg =
             "Processed"
@@ -96,12 +109,12 @@ instance Pretty LastSyncLog where
               <+> pretty numRollBackwards
               <+> "rollbacks in the last"
               <+> pretty (formatTime defaultTimeLocale "%s" timeSinceLastMsg)
-              <> "s"
+                <> "s"
        in case (timeSinceLastMsgM, cp, nt) of
             (Nothing, _, _) ->
               "Starting from"
                 <+> pretty cp
-                <> "."
+                  <> "."
                 <+> currentTipMsg timeSinceLastMsgM
             (Just _, _, C.ChainTipAtGenesis) ->
               "Not syncing. Node tip is at Genesis"
@@ -110,7 +123,7 @@ instance Pretty LastSyncLog where
               "Synchronising (0%)."
                 <+> currentTipMsg timeSinceLastMsgM
                 <+> processingSummaryMsg timeSinceLastMsg
-                <> "."
+                  <> "."
             ( Just timeSinceLastMsg
               , C.ChainPoint (C.SlotNo chainSyncSlot) _
               , C.ChainTip (C.SlotNo nodeTipSlot) _ _
@@ -119,7 +132,7 @@ instance Pretty LastSyncLog where
                     "Fully synchronised."
                       <+> currentTipMsg timeSinceLastMsgM
                       <+> processingSummaryMsg timeSinceLastMsg
-                      <> "."
+                        <> "."
             ( Just timeSinceLastMsg
               , C.ChainPoint (C.SlotNo chainSyncSlot) _
               , C.ChainTip (C.SlotNo nodeTipSlot) _ _

marconi-chain-index/src/Marconi/ChainIndex/Node/Client/Retry.hs

@@ -33,6 +33,9 @@ data RetryState = RetryState
   , secondsBeforeNextRetry :: !Word64
   }
 
+-- TODO: this function is problematic because it is used both inside the
+-- ChainIndexT context and outside to initialize the config
+-- see Marconi.ChainIndex.Run.run
 withNodeConnectRetry :: forall a. MarconiTrace IO -> RetryConfig -> FilePath -> IO a -> IO a
 withNodeConnectRetry stdoutTrace retryConfig socketPath action = do
   let initialRetryState = RetryState 0 (baseTimeBeforeNextRetry retryConfig)
@@ -66,7 +69,7 @@ withNodeConnectRetry stdoutTrace retryConfig socketPath action = do
               <+> pretty socketPath
               <+> "does not exist. Retrying in"
               <+> pretty (secondsBeforeNextRetry retryState)
-              <> "s ..."
+                <> "s ..."
 
           threadDelay $ fromIntegral $ secondsBeforeNextRetry retryState * 1_000_000
           runActionWithRetries

marconi-chain-index/src/Marconi/ChainIndex/Run.hs

@@ -16,6 +16,7 @@ import Marconi.ChainIndex.Node.Client.Retry (withNodeConnectRetry)
 import Marconi.ChainIndex.Types (
   RunIndexerConfig (RunIndexerConfig),
   UtxoIndexerConfig (UtxoIndexerConfig),
+  runChainIndexerT,
  )
 import Marconi.ChainIndex.Utils qualified as Utils
 import System.Directory (createDirectoryIfMissing)
@@ -62,15 +63,15 @@ run = do
     securityParam <- withNodeConnectRetry marconiTrace retryConfig socketPath $ do
       Utils.toException $ Utils.querySecurityParam @Void networkId socketPath
 
-    Indexers.runIndexers
+    runChainIndexerT
+      (Indexers.runIndexers indexers)
       ( RunIndexerConfig
           marconiTrace
           retryConfig
           securityParam
           networkId
           (Cli.optionsChainPoint $ Cli.commonOptions o)
           socketPath
+          (Cli.optionsMinIndexingDepth $ Cli.commonOptions o)
+          (Cli.optionsFailsIfResync o)
       )
-      (Cli.optionsMinIndexingDepth $ Cli.commonOptions o)
-      (Cli.optionsFailsIfResync o)
-      indexers

marconi-chain-index/src/Marconi/ChainIndex/Types.hs

@@ -6,6 +6,9 @@
 
 -- | This module provides several type aliases and utility functions to deal with them.
 module Marconi.ChainIndex.Types (
+  ChainIndexerT (..),
+  runChainIndexerT,
+
   -- * Config for retrying
   RetryConfig (..),
 
@@ -50,12 +53,17 @@ module Marconi.ChainIndex.Types (
   runIndexerConfigNetworkId,
   runIndexerConfigChainPoint,
   runIndexerConfigSocketPath,
+  runIndexerConfigIndexingDepth,
+  runIndexerConfigShouldFailIfResync,
 ) where
 
 import Cardano.Api qualified as C
 import Cardano.Api.Extended.Streaming (BlockEvent (BlockEvent, blockInMode, blockTime, epochNo))
 import Cardano.BM.Data.Trace (Trace)
 import Control.Lens.TH qualified as Lens
+import Control.Monad.Reader (MonadReader)
+import Control.Monad.Trans (MonadTrans (lift))
+import Control.Monad.Trans.Reader (ReaderT (runReaderT))
 import Data.Aeson (FromJSON, ToJSON)
 import Data.Aeson qualified as Aeson
 import Data.Set.NonEmpty (NESet)
@@ -65,6 +73,17 @@ import Database.SQLite.Simple.ToField qualified as SQL
 import GHC.Generics (Generic)
 import Prettyprinter (Doc)
 
+newtype ChainIndexerT m a = ChainIndexerT
+  {getChainIndexerT :: ReaderT (RunIndexerConfig m) m a}
+  deriving newtype (Functor, Applicative, Monad, MonadReader (RunIndexerConfig m))
+
+instance MonadTrans ChainIndexerT where
+  lift = ChainIndexerT . lift
+
+runChainIndexerT :: ChainIndexerT m a -> RunIndexerConfig m -> m a
+runChainIndexerT (ChainIndexerT chainIndexerT) config =
+  runReaderT chainIndexerT config
+
 -- | Config type for node retries
 data RetryConfig = RetryConfig
   { baseTimeBeforeNextRetry :: !Word64
@@ -142,13 +161,15 @@ newtype TxIndexInBlock = TxIndexInBlock Word64
     )
 
 -- | Common configuration required to run indexers
-data RunIndexerConfig = RunIndexerConfig
-  { _runIndexerConfigTrace :: MarconiTrace IO
+data RunIndexerConfig m = RunIndexerConfig
+  { _runIndexerConfigTrace :: MarconiTrace m
   , _runIndexerConfigRetryConfig :: RetryConfig
   , _runIndexerConfigSecurityParam :: SecurityParam
   , _runIndexerConfigNetworkId :: C.NetworkId
   , _runIndexerConfigChainPoint :: C.ChainPoint
   , _runIndexerConfigSocketPath :: FilePath
+  , _runIndexerConfigIndexingDepth :: IndexingDepth
+  , _runIndexerConfigShouldFailIfResync :: ShouldFailIfResync
   }
 
 Lens.makeLenses ''RunIndexerConfig