Initial mini-protocols for Peras

cabal.project

@@ -18,7 +18,7 @@ index-state:
   , hackage.haskell.org 2025-08-05T15:28:56Z
 
   -- Bump this if you need newer packages from CHaP
-  , cardano-haskell-packages 2025-03-18T17:41:11Z
+  , cardano-haskell-packages 2025-09-11T16:20:37Z
 
 packages: ./cardano-ping
           ./monoidal-synchronisation

cardano-diffusion/api/lib/Cardano/Network/NodeToNode/Version.hs

@@ -10,13 +10,18 @@ module Cardano.Network.NodeToNode.Version
   , ConnectionMode (..)
   , nodeToNodeVersionCodec
   , nodeToNodeCodecCBORTerm
+    -- * Feature predicates
+  , isPerasEnabled
   ) where
 
+import Data.Set (Set)
+import Data.Set qualified as Set
 import Data.Text (Text)
 import Data.Text qualified as T
 
 import Codec.CBOR.Term qualified as CBOR
 
+import Cardano.Base.FeatureFlags
 import Control.DeepSeq
 import GHC.Generics
 import NoThunks.Class (NoThunks)
@@ -72,16 +77,22 @@ data NodeToNodeVersion =
     -- ^ Plomin HF, mandatory on mainnet as of 2025.01.29
   | NodeToNodeV_15
     -- ^ SRV support
+  | NodeToNodeV_16
+    -- ^ Experimental.
+    --
+    -- Adds Peras mini-protocols (if 'PerasFlag' is set).
   deriving (Eq, Ord, Enum, Bounded, Show, Generic, NFData, NoThunks)
 
 nodeToNodeVersionCodec :: CodecCBORTerm (Text, Maybe Int) NodeToNodeVersion
 nodeToNodeVersionCodec = CodecCBORTerm { encodeTerm, decodeTerm }
   where
     encodeTerm NodeToNodeV_14 = CBOR.TInt 14
     encodeTerm NodeToNodeV_15 = CBOR.TInt 15
+    encodeTerm NodeToNodeV_16 = CBOR.TInt 16
 
     decodeTerm (CBOR.TInt 14) = Right NodeToNodeV_14
     decodeTerm (CBOR.TInt 15) = Right NodeToNodeV_15
+    decodeTerm (CBOR.TInt 16) = Right NodeToNodeV_16
     decodeTerm (CBOR.TInt n) = Left ( T.pack "decode NodeToNodeVersion: unknown tag: "
                                         <> T.pack (show n)
                                     , Just n
@@ -128,6 +139,7 @@ nodeToNodeCodecCBORTerm =
     \case
       NodeToNodeV_14 -> codec
       NodeToNodeV_15 -> codec
+      NodeToNodeV_16 -> codec
   where
     codec = CodecCBORTerm { encodeTerm = encodeTerm, decodeTerm = decodeTerm }
 
@@ -170,3 +182,8 @@ nodeToNodeCodecCBORTerm =
 
 
 data ConnectionMode = UnidirectionalMode | DuplexMode
+
+isPerasEnabled :: Set CardanoFeatureFlag -> NodeToNodeVersion -> Bool
+isPerasEnabled featureFlags v =
+       Set.member PerasFlag featureFlags
+    && v >= NodeToNodeV_16

cardano-diffusion/cardano-diffusion.cabal

@@ -68,7 +68,9 @@ library api
   build-depends:
     aeson,
     base >=4.14 && <4.22,
+    cardano-base ^>=0.1,
     cborg >=0.2.1 && <0.3,
+    containers,
     deepseq,
     nothunks,
     ouroboros-network:api,
@@ -151,6 +153,7 @@ library
     aeson,
     base >=4.14 && <4.22,
     bytestring,
+    cardano-base ^>=0.1,
     cardano-diffusion:{api, protocols},
     containers,
     contra-tracer,

cardano-diffusion/changelog.d/20250929_093633_alexander.esgen_peras_diffusion.md

@@ -0,0 +1,15 @@
+### Breaking
+
+- Added `NodeToNodeV_16`.
+
+- `Cardano.Network.NodeToNode`: Added support for certificate and vote diffusion mini-protocols for Ouroboros Peras.
+  - New fields in `NodeToNodeProtocols`.
+  - Protocol limits `perasCertDiffusionProtocolLimits`/`perasCertVoteProtocolLimits`.
+  - Mini-protocol numbers `perasCertDiffusionMiniProtocolNum`/`perasVoteDiffusionMiniProtocolNum`.
+  - `nodeToNodeProtocols` now takes a new `Set FeatureFlag` argument to control experimental features like Ouroboros Peras.
+
+### Non-Breaking
+
+- Added `isPerasEnabled` predicate.
+
+- `Cardano.Network.OrphanInstances`: Added support for `NodeToNodeV_16`.

cardano-diffusion/lib/Cardano/Network/NodeToNode.hs

@@ -21,6 +21,8 @@ module Cardano.Network.NodeToNode
   , txSubmissionProtocolLimits
   , keepAliveProtocolLimits
   , peerSharingProtocolLimits
+  , perasCertDiffusionProtocolLimits
+  , perasVoteDiffusionProtocolLimits
   , defaultMiniProtocolParameters
   , NodeToNodeVersion (..)
   , NodeToNodeVersionData (..)
@@ -69,17 +71,22 @@ module Cardano.Network.NodeToNode
   , txSubmissionMiniProtocolNum
   , keepAliveMiniProtocolNum
   , peerSharingMiniProtocolNum
+  , perasCertDiffusionMiniProtocolNum
+  , perasVoteDiffusionMiniProtocolNum
   ) where
 
 import Control.Exception (SomeException)
 
 import Data.ByteString.Lazy qualified as BL
+import Data.Set (Set)
 import Data.Word
 
 import Network.Mux qualified as Mx
 import Network.Socket (Socket, StructLinger (..))
 import Network.Socket qualified as Socket
 
+import Cardano.Base.FeatureFlags (CardanoFeatureFlag (..))
+
 import Cardano.Network.NodeToNode.Version
 import Cardano.Network.Protocol.Handshake.Codec
 
@@ -96,6 +103,7 @@ import Ouroboros.Network.PeerSelection.PeerAdvertise (PeerAdvertise (..))
 import Ouroboros.Network.PeerSelection.PeerSharing (PeerSharing (..))
 import Ouroboros.Network.Protocol.Handshake.Type
 import Ouroboros.Network.Protocol.Handshake.Version hiding (Accept)
+import Ouroboros.Network.Protocol.ObjectDiffusion.Type (NumObjectsOutstanding)
 import Ouroboros.Network.Protocol.TxSubmission2.Type (NumTxIdsToAck (..))
 import Ouroboros.Network.Server.RateLimiting
 import Ouroboros.Network.SizeInBytes
@@ -108,25 +116,32 @@ import Ouroboros.Network.TxSubmission.Inbound.V2.Policy (TxDecisionPolicy (..),
 data NodeToNodeProtocols appType initiatorCtx responderCtx bytes m a b = NodeToNodeProtocols {
     -- | chain-sync mini-protocol
     --
-    chainSyncProtocol    :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
+    chainSyncProtocol          :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
 
     -- | block-fetch mini-protocol
     --
-    blockFetchProtocol   :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
+    blockFetchProtocol         :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
 
     -- | tx-submission mini-protocol
     --
-    txSubmissionProtocol :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
+    txSubmissionProtocol       :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
+
+    -- | Peras certificate diffusion mini-protocol
+    --
+    perasCertDiffusionProtocol :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
+
+    -- | Peras vote diffusion mini-protocol
+    --
+    perasVoteDiffusionProtocol :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
 
     -- | keep-alive mini-protocol
     --
-    keepAliveProtocol    :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
+    keepAliveProtocol          :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b,
 
     -- | peer sharing mini-protocol
     --
-    peerSharingProtocol  :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b
-
-  }
+    peerSharingProtocol        :: RunMiniProtocol appType initiatorCtx responderCtx bytes m a b
+}
 
 type NodeToNodeProtocolsWithExpandedCtx appType ntnAddr bytes m a b =
     NodeToNodeProtocols appType (ExpandedInitiatorContext ntnAddr m) (ResponderContext ntnAddr) bytes m a b
@@ -135,11 +150,11 @@ type NodeToNodeProtocolsWithMinimalCtx  appType ntnAddr bytes m a b =
 
 
 data MiniProtocolParameters = MiniProtocolParameters {
-      chainSyncPipeliningHighMark :: !Word16,
+      chainSyncPipeliningHighMark     :: !Word16,
       -- ^ high threshold for pipelining (we will never exceed that many
       -- messages pipelined).
 
-      chainSyncPipeliningLowMark  :: !Word16,
+      chainSyncPipeliningLowMark      :: !Word16,
       -- ^ low threshold: if we hit the 'chainSyncPipeliningHighMark' we will
       -- listen for responses until there are at most
       -- 'chainSyncPipeliningLowMark' pipelined message
@@ -149,19 +164,51 @@ data MiniProtocolParameters = MiniProtocolParameters {
       -- Note: 'chainSyncPipeliningLowMark' and 'chainSyncPipeliningLowMark'
       -- are passed to 'pipelineDecisionLowHighMark'.
 
-      blockFetchPipeliningMax     :: !Word16,
+      blockFetchPipeliningMax         :: !Word16,
       -- ^ maximal number of pipelined messages in 'block-fetch' mini-protocol.
 
-      txDecisionPolicy            :: !TxDecisionPolicy
+      txDecisionPolicy                :: !TxDecisionPolicy,
       -- ^ tx submission protocol decision logic parameters
+
+      perasCertDiffusionMaxFifoLength :: !NumObjectsOutstanding,
+      -- ^ Maximum number of PerasCerts in the outbound peer's outstanding FIFO.
+      --
+      -- This indirectly limits the number of pipelined requests from the inbound peer:
+      -- the inbound peer can only request @n@ new IDs if the execution of preceding
+      -- requests would result in at least @n@ empty seats in the FIFO.
+      --
+      -- In the worst case:
+      --
+      --   * The inbound peer requests IDs and objects one by one.
+      --   * The inbound peer is aware of @perasCertDiffusionMaxFifoLength@ IDs for objects
+      --     it hasn't requested yet (i.e., the FIFO is full).
+      --
+      -- Then, the inbound peer can pipeline at most @perasCertDiffusionMaxFifoLength@
+      -- requests for one object each (with a known ID), and up to
+      -- @perasCertDiffusionMaxFifoLength@ requests for one new ID each.
+      --
+      -- So, the theoretical maximum pipeline size is
+      -- @2 * perasCertDiffusionMaxFifoLength@, but in practice the pipeline size will
+      -- be much smaller, as the inbound peer typically batches requests.
+
+      perasVoteDiffusionMaxFifoLength :: !NumObjectsOutstanding
+      -- ^ Maximum number of PerasVotes in the outbound peer's outstanding FIFO.
+      -- See comment on 'perasCertDiffusionMaxFifoLength' for more details to
+      -- understand why this indirectly limits the number of pipelined requests.
     }
 
 defaultMiniProtocolParameters :: MiniProtocolParameters
 defaultMiniProtocolParameters = MiniProtocolParameters {
-      chainSyncPipeliningLowMark  = 200
-    , chainSyncPipeliningHighMark = 300
-    , blockFetchPipeliningMax     = 100
-    , txDecisionPolicy            = defaultTxDecisionPolicy
+      chainSyncPipeliningLowMark      = 200
+    , chainSyncPipeliningHighMark     = 300
+    , blockFetchPipeliningMax         = 100
+    , txDecisionPolicy                = defaultTxDecisionPolicy
+    -- | TODO: this value is still being discussed.
+    -- See https://github.com/tweag/cardano-peras/issues/97 for reference.
+    , perasCertDiffusionMaxFifoLength = 10
+    -- | TODO: this value is still being discussed.
+    -- See https://github.com/tweag/cardano-peras/issues/97 for reference.
+    , perasVoteDiffusionMaxFifoLength = 10_000
   }
 
 -- | Make an 'OuroborosApplication' for the bundle of mini-protocols that
@@ -183,23 +230,26 @@ defaultMiniProtocolParameters = MiniProtocolParameters {
 -- both protocols, e.g.  wireshark plugins.
 --
 nodeToNodeProtocols
-  :: MiniProtocolParameters
+  :: Set CardanoFeatureFlag
+  -> MiniProtocolParameters
   -> NodeToNodeProtocols muxMode initiatorCtx responderCtx bytes m a b
   -> NodeToNodeVersion
   -- ^ negotiated version number
   -> NodeToNodeVersionData
   -- ^ negotiated version data
   -> OuroborosBundle muxMode initiatorCtx responderCtx bytes m a b
-nodeToNodeProtocols miniProtocolParameters protocols
-                    _version NodeToNodeVersionData { peerSharing }
+nodeToNodeProtocols featureFlags miniProtocolParameters protocols
+                    version NodeToNodeVersionData { peerSharing }
                     =
     TemperatureBundle
       -- Hot protocols: 'chain-sync', 'block-fetch' and 'tx-submission'.
       (WithHot $
         case protocols of
           NodeToNodeProtocols { chainSyncProtocol,
                                 blockFetchProtocol,
-                                txSubmissionProtocol
+                                txSubmissionProtocol,
+                                perasCertDiffusionProtocol,
+                                perasVoteDiffusionProtocol
                               } ->
             [ MiniProtocol {
                 miniProtocolNum    = chainSyncMiniProtocolNum,
@@ -219,7 +269,23 @@ nodeToNodeProtocols miniProtocolParameters protocols
                 miniProtocolLimits = txSubmissionProtocolLimits miniProtocolParameters,
                 miniProtocolRun    = txSubmissionProtocol
               }
-            ])
+            ]
+              <> concat [perasMiniProtocols | isPerasEnabled featureFlags version]
+           where
+             perasMiniProtocols =
+               [ MiniProtocol {
+                   miniProtocolNum    = perasCertDiffusionMiniProtocolNum,
+                   miniProtocolStart  = StartOnDemand,
+                   miniProtocolLimits = perasCertDiffusionProtocolLimits miniProtocolParameters,
+                   miniProtocolRun    = perasCertDiffusionProtocol
+                 }
+               , MiniProtocol {
+                   miniProtocolNum    = perasVoteDiffusionMiniProtocolNum,
+                   miniProtocolStart  = StartOnDemand,
+                   miniProtocolLimits = perasVoteDiffusionProtocolLimits miniProtocolParameters,
+                   miniProtocolRun    = perasVoteDiffusionProtocol
+                 }
+               ])
 
       -- Warm protocols: reserved for 'tip-sample'.
       (WithWarm [])
@@ -255,7 +321,9 @@ chainSyncProtocolLimits
   , blockFetchProtocolLimits
   , txSubmissionProtocolLimits
   , keepAliveProtocolLimits
-  , peerSharingProtocolLimits :: MiniProtocolParameters -> MiniProtocolLimits
+  , peerSharingProtocolLimits
+  , perasCertDiffusionProtocolLimits
+  , perasVoteDiffusionProtocolLimits :: MiniProtocolParameters -> MiniProtocolLimits
 
 chainSyncProtocolLimits MiniProtocolParameters { chainSyncPipeliningHighMark } =
   MiniProtocolLimits {
@@ -374,6 +442,26 @@ peerSharingProtocolLimits _ =
   maximumIngressQueue = 4 * 1440
   }
 
+perasCertDiffusionProtocolLimits MiniProtocolParameters { perasCertDiffusionMaxFifoLength } =
+  MiniProtocolLimits {
+      -- The reasoning here is very similar to the 'txSubmissionProtocolLimits'.
+      --
+      -- Peras certificates will definitely be smaller than 20 kB; potentially
+      -- even much smaller.
+      -- See https://github.com/tweag/cardano-peras/issues/97
+      maximumIngressQueue = addSafetyMargin $
+        fromIntegral perasCertDiffusionMaxFifoLength * 20_000
+    }
+
+perasVoteDiffusionProtocolLimits MiniProtocolParameters { perasVoteDiffusionMaxFifoLength } =
+  MiniProtocolLimits {
+      -- Peras votes are expected to be much smaller than Peras certificates.
+      -- We assume an upper bound of 1 kB per vote.
+      -- See https://github.com/tweag/cardano-peras/issues/97
+      maximumIngressQueue = addSafetyMargin $
+        fromIntegral perasVoteDiffusionMaxFifoLength * 1_000
+    }
+
 chainSyncMiniProtocolNum :: MiniProtocolNum
 chainSyncMiniProtocolNum = MiniProtocolNum 2
 
@@ -389,6 +477,12 @@ keepAliveMiniProtocolNum = MiniProtocolNum 8
 peerSharingMiniProtocolNum :: MiniProtocolNum
 peerSharingMiniProtocolNum = MiniProtocolNum 10
 
+perasCertDiffusionMiniProtocolNum :: MiniProtocolNum
+perasCertDiffusionMiniProtocolNum = MiniProtocolNum 16
+
+perasVoteDiffusionMiniProtocolNum :: MiniProtocolNum
+perasVoteDiffusionMiniProtocolNum = MiniProtocolNum 17
+
 -- | A specialised version of @'Ouroboros.Network.Socket.connectToNode'@.
 --
 connectTo

cardano-diffusion/orphan-instances/Cardano/Network/OrphanInstances.hs

@@ -67,12 +67,14 @@ instance FromJSON NodeToNodeVersion where
   parseJSON = \case
     Number 14 -> pure NodeToNodeV_14
     Number 15 -> pure NodeToNodeV_15
+    Number 16 -> pure NodeToNodeV_16
     Number x  -> fail $ "FromJSON.NodeToNodeVersion: unsupported node-to-node protocol version " ++ show x
     x         -> fail $ "FromJSON.NodeToNodeVersion: error parsing NodeToNodeVersion: " ++ show x
 
 instance ToJSON NodeToNodeVersion where
   toJSON NodeToNodeV_14 = Number 14
   toJSON NodeToNodeV_15 = Number 15
+  toJSON NodeToNodeV_16 = Number 16
 
 instance FromJSON NodeToClientVersion where
   parseJSON = \case

cardano-diffusion/protocols/cddl/specs/handshake-node-to-node-v14.cddl

@@ -15,7 +15,7 @@ msgQueryReply      = [3, versionTable]
 ; The codec only accepts definite-length maps.
 versionTable = { * versionNumber_v14 => v14.nodeToNodeVersionData }
 
-versionNumber_v14 = 14 / 15
+versionNumber_v14 = 14 / 15 / 16
 
 ; All version numbers
 versionNumbers = versionNumber_v14