Implement and wire-in the ObjectDiffusion miniprotocol for Peras certificate diffusion

cabal.project

@@ -59,9 +59,11 @@ allow-newer:
 source-repository-package
    type: git
    location: https://github.com/IntersectMBO/ouroboros-network
-   tag: b07a86ed853b63881b5a83e57508902f1562ac01
-   --sha256: sha256-n/XX0+cQegq2a1cAfmGx30T64eix4oEXzpVEFCKqmg0=
+   tag: c2e936f454a0026b9a854e5f230714de81b9965c
+   --sha256: sha256-139VtT1VJkBqIcqf+vak7h4Fh+Z748dHoHwaCCpKOy4=
    subdir:
+     ouroboros-network
+     ouroboros-network-protocols
      ouroboros-network-api
      ouroboros-network
 

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
     }
 
 {-------------------------------------------------------------------------------

ouroboros-consensus-diffusion/src/unstable-diffusion-testlib/Test/ThreadNet/Network.hs

@@ -123,8 +123,8 @@ import Ouroboros.Network.NodeToNode
   ( ConnectionId (..)
   , ExpandedInitiatorContext (..)
   , IsBigLedgerPeer (..)
-  , MiniProtocolParameters (..)
   , ResponderContext (..)
+  , defaultMiniProtocolParameters
   )
 import Ouroboros.Network.PeerSelection.Governor
   ( makePublicPeerSelectionStateVar
@@ -1056,13 +1056,7 @@ runThreadNetwork
               , mempoolCapacityOverride = NoMempoolCapacityBytesOverride
               , keepAliveRng = kaRng
               , peerSharingRng = psRng
-              , miniProtocolParameters =
-                  MiniProtocolParameters
-                    { chainSyncPipeliningHighMark = 4
-                    , chainSyncPipeliningLowMark = 2
-                    , blockFetchPipeliningMax = 10
-                    , txSubmissionMaxUnacked = 1000 -- TODO ?
-                    }
+              , miniProtocolParameters = defaultMiniProtocolParameters
               , blockFetchConfiguration =
                   BlockFetchConfiguration
                     { bfcMaxConcurrencyBulkSync = 1

ouroboros-consensus/ouroboros-consensus.cabal

@@ -183,6 +183,11 @@ library
     Ouroboros.Consensus.MiniProtocol.LocalStateQuery.Server
     Ouroboros.Consensus.MiniProtocol.LocalTxMonitor.Server
     Ouroboros.Consensus.MiniProtocol.LocalTxSubmission.Server
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.PerasCert
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Outbound
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.PerasCert
     Ouroboros.Consensus.Node.GsmState
     Ouroboros.Consensus.Node.InitStorage
     Ouroboros.Consensus.Node.NetworkProtocolVersion
@@ -593,6 +598,8 @@ test-suite consensus-test
     Test.Consensus.MiniProtocol.ChainSync.CSJ
     Test.Consensus.MiniProtocol.ChainSync.Client
     Test.Consensus.MiniProtocol.LocalStateQuery.Server
+    Test.Consensus.MiniProtocol.ObjectDiffusion.PerasCert.Smoke
+    Test.Consensus.MiniProtocol.ObjectDiffusion.Smoke
     Test.Consensus.Peras.WeightSnapshot
     Test.Consensus.Util.MonadSTM.NormalForm
     Test.Consensus.Util.Versioned