Wire-in the PerasCertDiffusion protocol in NodeToNode

Author: tbagrel1

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs

@@ -68,6 +68,10 @@ import Ouroboros.Consensus.MiniProtocol.ChainSync.Client
   )
 import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client as CsClient
 import Ouroboros.Consensus.MiniProtocol.ChainSync.Server
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound (objectDiffusionInbound)
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.PerasCert
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Outbound (objectDiffusionOutbound)
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.PerasCert
 import Ouroboros.Consensus.Node.ExitPolicy
 import Ouroboros.Consensus.Node.NetworkProtocolVersion
 import Ouroboros.Consensus.Node.Run
@@ -197,6 +201,19 @@ data Handlers m addr blk = Handlers
       NodeToNodeVersion ->
       ConnectionId addr ->
       TxSubmissionServerPipelined (GenTxId blk) (GenTx blk) m ()
+  , hPerasCertDiffusionInbound ::
+      NodeToNodeVersion ->
+      ConnectionId addr ->
+      PerasCertDiffusionInboundPipelined blk m ()
+  -- ^ TODO: We should pass 'hPerasCertDiffusionInbound' to the network
+  -- layer, as per https://github.com/tweag/cardano-peras/issues/78
+  , hPerasCertDiffusionOutbound ::
+      NodeToNodeVersion ->
+      ControlMessageSTM m ->
+      ConnectionId addr ->
+      PerasCertDiffusionOutbound blk m ()
+  -- ^ TODO: We should pass 'hPerasCertDiffusionOutbound' to the network
+  -- layer, as per https://github.com/tweag/cardano-peras/issues/78
   , hKeepAliveClient ::
       NodeToNodeVersion ->
       ControlMessageSTM m ->
@@ -293,6 +310,22 @@ mkHandlers
             (mapTxSubmissionMempoolReader txForgetValidated $ getMempoolReader getMempool)
             (getMempoolWriter getMempool)
             version
+      , hPerasCertDiffusionInbound = \version peer ->
+          objectDiffusionInbound
+            (contramap (TraceLabelPeer peer) (Node.perasCertDiffusionInboundTracer tracers))
+            ( perasCertDiffusionMaxFifoLength miniProtocolParameters
+            , 10 -- TODO https://github.com/tweag/cardano-peras/issues/97
+            , 10 -- TODO https://github.com/tweag/cardano-peras/issues/97
+            )
+            (makePerasCertPoolWriterFromChainDB $ getChainDB)
+            version
+      , hPerasCertDiffusionOutbound = \version controlMessageSTM peer ->
+          objectDiffusionOutbound
+            (contramap (TraceLabelPeer peer) (Node.perasCertDiffusionOutboundTracer tracers))
+            (perasCertDiffusionMaxFifoLength miniProtocolParameters)
+            (makePerasCertPoolReaderFromChainDB $ getChainDB)
+            version
+            controlMessageSTM
       , hKeepAliveClient = \_version -> keepAliveClient (Node.keepAliveClientTracer tracers) keepAliveRng
       , hKeepAliveServer = \_version _peer -> keepAliveServer
       , hPeerSharingClient = \_version controlMessageSTM _peer -> peerSharingClient controlMessageSTM

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node/Tracers.hs

@@ -42,6 +42,7 @@ import Ouroboros.Consensus.MiniProtocol.ChainSync.Server
 import Ouroboros.Consensus.MiniProtocol.LocalTxSubmission.Server
   ( TraceLocalTxSubmissionServerEvent (..)
   )
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.PerasCert
 import Ouroboros.Consensus.Node.GSM (TraceGsmEvent)
 import Ouroboros.Network.Block (Tip)
 import Ouroboros.Network.BlockFetch
@@ -87,6 +88,10 @@ data Tracers' remotePeer localPeer blk f = Tracers
   , csjTracer ::
       f (TraceLabelPeer remotePeer (CSJumping.TraceEventCsj remotePeer blk))
   , dbfTracer :: f (CSJumping.TraceEventDbf remotePeer)
+  , perasCertDiffusionInboundTracer ::
+      f (TraceLabelPeer remotePeer (TracePerasCertDiffusionInbound blk))
+  , perasCertDiffusionOutboundTracer ::
+      f (TraceLabelPeer remotePeer (TracePerasCertDiffusionOutbound blk))
   }
 
 instance
@@ -115,6 +120,8 @@ instance
       , gddTracer = f gddTracer
       , csjTracer = f csjTracer
       , dbfTracer = f dbfTracer
+      , perasCertDiffusionInboundTracer = f perasCertDiffusionInboundTracer
+      , perasCertDiffusionOutboundTracer = f perasCertDiffusionOutboundTracer
       }
    where
     f ::
@@ -151,6 +158,8 @@ nullTracers =
     , gddTracer = nullTracer
     , csjTracer = nullTracer
     , dbfTracer = nullTracer
+    , perasCertDiffusionInboundTracer = nullTracer
+    , perasCertDiffusionOutboundTracer = nullTracer
     }
 
 showTracers ::
@@ -189,6 +198,8 @@ showTracers tr =
     , gddTracer = showTracing tr
     , csjTracer = showTracing tr
     , dbfTracer = showTracing tr
+    , perasCertDiffusionInboundTracer = showTracing tr
+    , perasCertDiffusionOutboundTracer = showTracing tr
     }
 
 {-------------------------------------------------------------------------------