Allways close chainsync node2client reader state

IntersectMBO · Jan 18, 2021 · 3d66c2e · 3d66c2e
1 parent cda0651
commit 3d66c2e
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 20 deletions.
diff --git a/ouroboros-consensus/src/Ouroboros/Consensus/MiniProtocol/ChainSync/Server.hs b/ouroboros-consensus/src/Ouroboros/Consensus/MiniProtocol/ChainSync/Server.hs
@@ -8,6 +8,7 @@ module Ouroboros.Consensus.MiniProtocol.ChainSync.Server
   ( chainSyncHeadersServer
   , chainSyncBlocksServer
   , chainSyncHeaderServerReader
+  , chainSyncBlockServerReader
   , Tip
     -- * Trace events
   , TraceChainSyncServerEvent (..)
@@ -37,6 +38,13 @@ chainSyncHeaderServerReader
     -> m (Reader m blk (WithPoint blk (SerialisedHeader blk)))
 chainSyncHeaderServerReader chainDB registry = ChainDB.newReader chainDB registry getSerialisedHeaderWithPoint
 
+chainSyncBlockServerReader
+    :: ChainDB m blk
+    -> ResourceRegistry m
+    -> m (Reader m blk (WithPoint blk (Serialised blk)))
+chainSyncBlockServerReader chainDB registry = ChainDB.newReader chainDB registry getSerialisedBlockWithPoint
+
+
 
 -- | Chain Sync Server for block headers for a given a 'ChainDB'.
 --
@@ -67,12 +75,11 @@ chainSyncBlocksServer
     :: forall m blk. (IOLike m, HasHeader (Header blk))
     => Tracer m (TraceChainSyncServerEvent blk)
     -> ChainDB m blk
-    -> ResourceRegistry m
+    -> Reader m blk (WithPoint blk (Serialised blk))
     -> ChainSyncServer (Serialised blk) (Point blk) (Tip blk) m ()
-chainSyncBlocksServer tracer chainDB registry =
-    ChainSyncServer $ do
-      rdr <- ChainDB.newReader chainDB registry getSerialisedBlockWithPoint
-      let ChainSyncServer server = chainSyncServerForReader tracer chainDB rdr
+chainSyncBlocksServer tracer chainDB rdr =
+    ChainSyncServer $
+      let ChainSyncServer server = chainSyncServerForReader tracer chainDB rdr in
       server
 
 -- | A chain sync server.

diff --git a/ouroboros-consensus/src/Ouroboros/Consensus/Network/NodeToClient.hs b/ouroboros-consensus/src/Ouroboros/Consensus/Network/NodeToClient.hs
@@ -86,7 +86,7 @@ import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
 -- | Protocol handlers for node-to-client (local) communication
 data Handlers m peer blk = Handlers {
       hChainSyncServer
-        :: ResourceRegistry m
+        :: ChainDB.Reader m blk (ChainDB.WithPoint blk (Serialised blk))
         -> ChainSyncServer (Serialised blk) (Point blk) (Tip blk) m ()
 
     , hTxSubmissionServer
@@ -326,7 +326,7 @@ data Apps m peer bCS bTX bSQ a = Apps {
 
 -- | Construct the 'NetworkApplication' for the node-to-client protocols
 mkApps
-  :: forall m peer blk e bCS bTX bSQ.
+  :: forall m remotePeer localPeer blk e bCS bTX bSQ.
      ( IOLike m
      , Exception e
      , ShowProxy blk
@@ -335,29 +335,34 @@ mkApps
      , ShowProxy (GenTx blk)
      , ShowQuery (Query blk)
      )
-  => Tracers m peer blk e
+  => NodeKernel m remotePeer localPeer blk
+  -> Tracers m localPeer blk e
   -> Codecs blk e m bCS bTX bSQ
-  -> Handlers m peer blk
-  -> Apps m peer bCS bTX bSQ ()
-mkApps Tracers {..} Codecs {..} Handlers {..} =
+  -> Handlers m localPeer blk
+  -> Apps m localPeer bCS bTX bSQ ()
+mkApps kernel Tracers {..} Codecs {..} Handlers {..} =
     Apps {..}
   where
     aChainSyncServer
-      :: peer
+      :: localPeer
       -> Channel m bCS
       -> m ((), Maybe bCS)
     aChainSyncServer them channel = do
       labelThisThread "LocalChainSyncServer"
       withRegistry $ \registry ->
-        runPeer
-          (contramap (TraceLabelPeer them) tChainSyncTracer)
-          cChainSyncCodec
-          channel
-          $ chainSyncServerPeer
-          $ hChainSyncServer registry
+        bracket
+          (chainSyncBlockServerReader (getChainDB kernel) registry)
+          ChainDB.readerClose
+          (\rdr -> runPeer
+            (contramap (TraceLabelPeer them) tChainSyncTracer)
+            cChainSyncCodec
+            channel
+            $ chainSyncServerPeer
+            $ hChainSyncServer rdr
+          )
 
     aTxSubmissionServer
-      :: peer
+      :: localPeer
       -> Channel m bTX
       -> m ((), Maybe bTX)
     aTxSubmissionServer them channel = do
@@ -369,7 +374,7 @@ mkApps Tracers {..} Codecs {..} Handlers {..} =
         (localTxSubmissionServerPeer (pure hTxSubmissionServer))
 
     aStateQueryServer
-      :: peer
+      :: localPeer
       -> Channel m bSQ
       -> m ((), Maybe bSQ)
     aStateQueryServer them channel = do

diff --git a/ouroboros-consensus/src/Ouroboros/Consensus/Node.hs b/ouroboros-consensus/src/Ouroboros/Consensus/Node.hs
@@ -330,6 +330,7 @@ runWith RunNodeArgs{..} LowLevelRunNodeArgs{..} =
       -> NTC.Apps m (ConnectionId addrNTC) ByteString ByteString ByteString ()
     mkNodeToClientApps nodeKernelArgs nodeKernel version =
         NTC.mkApps
+          nodeKernel
           rnTraceNTC
           (NTC.defaultCodecs codecConfig version)
           (NTC.mkHandlers nodeKernelArgs nodeKernel)