Refactor chain sync event consuming code

marconi-cardano-core/cardano-api-extended/src/Cardano/Api/Extended/Streaming.hs

@@ -3,7 +3,7 @@
 
 module Cardano.Api.Extended.Streaming (
   withChainSyncEventStream,
-  withChainSyncBlockEventStream,
+  mkChainSyncConnection,
   ChainSyncEvent (..),
   ChainSyncEventException (..),
 
@@ -14,6 +14,8 @@ module Cardano.Api.Extended.Streaming (
   blocks,
   blocksPipelined,
   ignoreRollbacks,
+  ChainSyncConnection (..),
+  connectToLocalNodeWithChainSyncClient,
 ) where
 
 import Cardano.Api qualified as C
@@ -111,24 +113,33 @@ data BlockEvent = BlockEvent
   }
   deriving (Show)
 
+-- | The data necessary for connecting to a client using the chain sync protocol.
+data ChainSyncConnection r = ChainSyncConnection
+  { eventStream :: Stream (Of (ChainSyncEvent BlockEvent)) IO r
+  -- ^ the stream of events coming from the client
+  , nodeConnectInfo :: C.LocalNodeConnectInfo C.CardanoMode
+  -- ^ information about the local node
+  , chainSyncClient :: C.ChainSyncClient (C.BlockInMode C.CardanoMode) C.ChainPoint C.ChainTip IO ()
+  -- ^ the actual client to connect to
+  }
+
 {- | Uses the chain-sync mini-protocol to connect to a locally running node and fetch blocks from the
     given starting point, along with their @EpochNo@ and creation time.
 -}
-withChainSyncBlockEventStream
+mkChainSyncConnection
   :: FilePath
   -- ^ Path to the node socket
   -> C.NetworkId
   -> [C.ChainPoint]
   -- ^ The point on the chain to start streaming from
-  -> (Stream (Of (ChainSyncEvent BlockEvent)) IO r -> IO b)
-  -- ^ The stream consumer
-  -> IO b
-withChainSyncBlockEventStream socketPath networkId points consumer =
+  -> IO (ChainSyncConnection r)
+mkChainSyncConnection socketPath networkId points =
   do
-    -- The chain-sync client runs in a different thread passing the blocks it
+    -- The chain-sync client should be run in a different thread passing the blocks it
     -- receives to the stream consumer through a MVar. The chain-sync client
-    -- thread and the stream consumer will each block on each other and stay
-    -- in lockstep.
+    -- thread and the stream consumer will stay in lockstep. Once a block is emitted
+    -- by the client thread it will be consumed by the consumer thread.
+    -- See 'Marconi.Cardano.Core.Runner.connectToChainSyncAndStream'.
     --
     -- NOTE: choosing a MVar is a tradeoff towards simplicity. In this case a
     -- (bounded) queue could perform better. Indeed a properly-sized buffer
@@ -195,14 +206,8 @@ withChainSyncBlockEventStream socketPath networkId points consumer =
         eventLoop history = takeMVar nextChainSyncEventVar >>= fmap Right . attachEpochAndTime history
 
     history <- askHistory
-    withAsync (connectToLocalNodeWithChainSyncClient localNodeConnectInfo client) $ \a -> do
-      -- Make sure all exceptions in the client thread are passed to the consumer thread
-      link a
-      -- Run the consumer
-      consumer $ S.unfoldr eventLoop history
-    -- Let's rethrow exceptions from the client thread unwrapped, so that the
-    -- consumer does not have to know anything about async
-    `catch` \(ExceptionInLinkedThread _ (SomeException e)) -> throw e
+    let stream = S.unfoldr eventLoop history
+    return $ ChainSyncConnection stream localNodeConnectInfo client
 
 connectToLocalNodeWithChainSyncClient
   :: C.LocalNodeConnectInfo C.CardanoMode

marconi-cardano-core/marconi-cardano-core.cabal

@@ -89,6 +89,7 @@ library
   ------------------------
   build-depends:
     , aeson
+    , async
     , base                 >=4.9 && <5
     , bytestring
     , cborg

marconi-cardano-core/src/Marconi/Cardano/Core/Runner.hs

@@ -35,14 +35,25 @@ module Marconi.Cardano.Core.Runner (
 import Cardano.Api.Extended qualified as C
 import Cardano.Api.Extended.Streaming (
   BlockEvent (BlockEvent),
+  ChainSyncConnection (ChainSyncConnection),
   ChainSyncEvent (RollBackward, RollForward),
   ChainSyncEventException (NoIntersectionFound),
-  withChainSyncBlockEventStream,
+  connectToLocalNodeWithChainSyncClient,
+  mkChainSyncConnection,
  )
 import Cardano.BM.Trace qualified as Trace
 import Control.Concurrent qualified as Concurrent
+import Control.Concurrent.Async (
+  ExceptionInLinkedThread (ExceptionInLinkedThread),
+  link,
+  withAsync,
+ )
 import Control.Concurrent.STM qualified as STM
-import Control.Exception (catch)
+import Control.Exception (
+  SomeException (SomeException),
+  catch,
+  throw,
+ )
 import Control.Lens ((^.))
 import Control.Lens qualified as Lens
 import Control.Monad (void)
@@ -66,7 +77,6 @@ import Marconi.Cardano.Core.Types (
 import Marconi.Core qualified as Core
 import Prettyprinter (pretty)
 import Prettyprinter qualified as PP
-import Streaming qualified as S
 import Streaming.Prelude qualified as S
 
 -- | Runner pre-processing
@@ -136,11 +146,11 @@ runIndexer
       cBox <- Concurrent.newMVar indexer
       let processEvent = eventProcessing ^. runIndexerPreprocessEvent
           runChainSyncStream =
-            withChainSyncBlockEventStream
+            mkChainSyncConnection
               socketPath
               networkId
               [startingPoint]
-              (mkEventStream processEvent eventQueue)
+              >>= connectToChainSyncAndStream processEvent eventQueue
           whenNoIntersectionFound NoIntersectionFound =
             Trace.logError trace $
               PP.pretty NoIntersectionFoundLog
@@ -178,14 +188,28 @@ getBlockNo :: C.BlockInMode C.CardanoMode -> C.BlockNo
 getBlockNo (C.BlockInMode block _eraInMode) =
   case C.getBlockHeader block of C.BlockHeader _ _ b -> b
 
--- | Event preprocessing, to ease the coordinator work
-mkEventStream
+{- | Creates a new thread which connects to the local node. In the current thread it
+receives events via the chain sync protocol. These events are preprocessed and written
+to a 'TBQueue'.
+-}
+connectToChainSyncAndStream
   :: (ChainSyncEvent BlockEvent -> [Core.ProcessedInput C.ChainPoint a])
   -> STM.TBQueue (Core.ProcessedInput C.ChainPoint a)
-  -> S.Stream (S.Of (ChainSyncEvent BlockEvent)) IO r
+  -> ChainSyncConnection r
   -> IO r
-mkEventStream processEvent q =
-  S.mapM_ $ STM.atomically . traverse_ (STM.writeTBQueue q) . processEvent
+connectToChainSyncAndStream
+  processEvent
+  queue
+  (ChainSyncConnection stream localNodeConnectInfo client) =
+    withAsync (connectToLocalNodeWithChainSyncClient localNodeConnectInfo client) $ \a ->
+      do
+        -- Make sure all exceptions in the client thread are passed to the consumer thread
+        link a
+        -- Process and consume the stream
+        S.mapM_ (STM.atomically . traverse_ (STM.writeTBQueue queue) . processEvent) stream
+        -- Let's rethrow exceptions from the client thread unwrapped, so that the
+        -- consumer does not have to know anything about async
+        `catch` \(ExceptionInLinkedThread _ (SomeException e)) -> throw e
 
 withDistanceAndTipPreprocessor
   :: RunIndexerEventPreprocessing TipAndBlock