Allways close chainsync servers reader state

IntersectMBO · Jan 18, 2021 · 9cc87ef · 9cc87ef
1 parent ca53ae7
commit 9cc87ef
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 16 deletions.
diff --git a/ouroboros-consensus/src/Ouroboros/Consensus/MiniProtocol/ChainSync/Server.hs b/ouroboros-consensus/src/Ouroboros/Consensus/MiniProtocol/ChainSync/Server.hs
@@ -7,6 +7,7 @@
 module Ouroboros.Consensus.MiniProtocol.ChainSync.Server
   ( chainSyncHeadersServer
   , chainSyncBlocksServer
+  , chainSyncHeaderServerReader
   , Tip
     -- * Trace events
   , TraceChainSyncServerEvent (..)
@@ -29,6 +30,14 @@ import           Ouroboros.Consensus.Node.NetworkProtocolVersion
 import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Consensus.Util.ResourceRegistry (ResourceRegistry)
 
+
+chainSyncHeaderServerReader
+    :: ChainDB m blk
+    -> ResourceRegistry m
+    -> m (Reader m blk (WithPoint blk (SerialisedHeader blk)))
+chainSyncHeaderServerReader chainDB registry = ChainDB.newReader chainDB registry getSerialisedHeaderWithPoint
+
+
 -- | Chain Sync Server for block headers for a given a 'ChainDB'.
 --
 -- The node-to-node protocol uses the chain sync mini-protocol with chain
@@ -41,13 +50,12 @@ chainSyncHeadersServer
        )
     => Tracer m (TraceChainSyncServerEvent blk)
     -> ChainDB m blk
+    -> Reader m blk (WithPoint blk (SerialisedHeader blk))
     -> NodeToNodeVersion
-    -> ResourceRegistry m
     -> ChainSyncServer (SerialisedHeader blk) (Point blk) (Tip blk) m ()
-chainSyncHeadersServer tracer chainDB _version registry =
-    ChainSyncServer $ do
-      rdr <- ChainDB.newReader chainDB registry getSerialisedHeaderWithPoint
-      let ChainSyncServer server = chainSyncServerForReader tracer chainDB rdr
+chainSyncHeadersServer tracer chainDB rdr _version =
+    ChainSyncServer $
+      let ChainSyncServer server = chainSyncServerForReader tracer chainDB rdr in
       server
 
 -- | Chain Sync Server for blocks for a given a 'ChainDB'.
@@ -92,7 +100,7 @@ chainSyncServerForReader tracer chainDB rdr =
     idle = ServerStIdle {
         recvMsgRequestNext   = handleRequestNext,
         recvMsgFindIntersect = handleFindIntersect,
-        recvMsgDoneClient    = ChainDB.readerClose rdr
+        recvMsgDoneClient    = pure ()
       }
 
     idle' :: ChainSyncServer b (Point blk) (Tip blk) m ()

diff --git a/ouroboros-consensus/src/Ouroboros/Consensus/Network/NodeToNode.hs b/ouroboros-consensus/src/Ouroboros/Consensus/Network/NodeToNode.hs
@@ -88,6 +88,7 @@ import           Ouroboros.Consensus.Node.Run
 import           Ouroboros.Consensus.Node.Serialisation
 import qualified Ouroboros.Consensus.Node.Tracers as Node
 import           Ouroboros.Consensus.NodeKernel
+import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
 import           Ouroboros.Consensus.Util (ShowProxy)
 import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Consensus.Util.Orphans ()
@@ -112,8 +113,8 @@ data Handlers m peer blk = Handlers {
         -- closure include these and not need to be explicit about them here.
 
     , hChainSyncServer
-        :: NodeToNodeVersion
-        -> ResourceRegistry m
+        :: ChainDB.Reader m blk (ChainDB.WithPoint blk (SerialisedHeader blk))
+        -> NodeToNodeVersion
         -> ChainSyncServer (SerialisedHeader blk) (Point blk) (Tip blk) m ()
 
     -- TODO block fetch client does not have GADT view of the handlers.
@@ -488,14 +489,18 @@ mkApps kernel Tracers {..} Codecs {..} genChainSyncTimeout Handlers {..} =
       labelThisThread "ChainSyncServer"
       withRegistry $ \registry -> do
         chainSyncTimeout <- genChainSyncTimeout
-        runPeerWithLimits
-          (contramap (TraceLabelPeer them) tChainSyncSerialisedTracer)
-          cChainSyncCodecSerialised
-          (byteLimitsChainSync (const 0)) -- TODO: Real Bytelimits, see #1727
-          (timeLimitsChainSync chainSyncTimeout)
-          channel
-          $ chainSyncServerPeer
-          $ hChainSyncServer version registry
+        bracket
+          (chainSyncHeaderServerReader (getChainDB kernel) registry)
+          ChainDB.readerClose
+          (\rdr -> runPeerWithLimits
+            (contramap (TraceLabelPeer them) tChainSyncSerialisedTracer)
+            cChainSyncCodecSerialised
+            (byteLimitsChainSync (const 0)) -- TODO: Real Bytelimits, see #1727
+            (timeLimitsChainSync chainSyncTimeout)
+            channel
+            $ chainSyncServerPeer
+            $ hChainSyncServer rdr version
+          )
 
     aBlockFetchClient
       :: NodeToNodeVersion