Refactor ObjectDiffusion protocol internals

Author: tbagrel1

cabal.project

@@ -59,8 +59,8 @@ allow-newer:
 source-repository-package
    type: git
    location: https://github.com/IntersectMBO/ouroboros-network
-   tag: bca44ca1314b5ac2e57f1843d5ab7b848c318c45
-   --sha256: sha256-EAYEhii5KO8VnyNenef6rquHFJdubp0xckQpAQhtFA4=
+   tag: c2e936f454a0026b9a854e5f230714de81b9965c
+   --sha256: sha256-139VtT1VJkBqIcqf+vak7h4Fh+Z748dHoHwaCCpKOy4=
    subdir:
      ouroboros-network
      ouroboros-network-protocols

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

@@ -309,14 +309,16 @@ mkHandlers
       , hPerasCertDiffusionInbound = \version peer ->
           objectDiffusionInbound
             (contramap (TraceLabelPeer peer) (Node.certDiffusionInboundTracer tracers))
-            (certDiffusionMaxUnacked miniProtocolParameters)
-            (makePerasCertPoolReaderFromChainDB $ getChainDB)
+            ( perasCertDiffusionMaxFifoLength miniProtocolParameters
+            , 10 -- TODO: change this to a sensible value
+            , 10 -- TODO: change this to a sensible value
+            )
             (makePerasCertPoolWriterFromChainDB $ getChainDB)
             version
       , hPerasCertDiffusionOutbound = \version controlMessageSTM peer ->
           objectDiffusionOutbound
             (contramap (TraceLabelPeer peer) (Node.certDiffusionOutboundTracer tracers))
-            (certDiffusionMaxUnacked miniProtocolParameters)
+            (perasCertDiffusionMaxFifoLength miniProtocolParameters)
             (makePerasCertPoolReaderFromChainDB $ getChainDB)
             version
             controlMessageSTM

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Inbound.hs

@@ -1,34 +1,29 @@
 module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
   ( ObjectPoolReader (..)
-  , ObjectPoolSnapshot (..)
   , ObjectPoolWriter (..)
   ) where
 
-import Control.Monad.Class.MonadSTM (STM)
-import Ouroboros.Network.SizeInBytes (SizeInBytes)
+import Control.Concurrent.Class.MonadSTM.Strict (STM)
+import Data.Word (Word64)
 
 data ObjectPoolReader objectId object ticketNo m
   = ObjectPoolReader
-  { rdrGetObjectId :: object -> objectId
+  { oprObjectId :: object -> objectId
   -- ^ Return the id of the specified object
-  , objectPoolGetSnapshot :: STM m (ObjectPoolSnapshot objectId object ticketNo)
-  -- ^ Get a snapshot of the object pool
-  , objectPoolZeroTicketNo :: ticketNo
-  -- ^ Ticket number before the first item in the pool (so objectPoolObjectsAfter objectPoolZeroTicketNo returns all possible objects)
-  }
-data ObjectPoolSnapshot objectId object ticketNo
-  = ObjectPoolSnapshot
-  -- TODO: revisit when we will have to load certificates from disk. This method might not be pure anymore
-  { objectPoolObjectsAfter :: ticketNo -> [(object, ticketNo, SizeInBytes)]
-  -- ^ Get all objects having a ticket number strictly greater than the given one, along with their ticket numbers and sizes
-  , objectPoolHasObject :: objectId -> Bool
-  -- ^ Check if the object pool contains an object with the given id
+  , oprZeroTicketNo :: ticketNo
+  -- ^ Ticket number before the first item in the pool (so oprObjectsAfter oprZeroTicketNo returns all possible objects)
+  , oprObjectsAfter :: ticketNo -> Word64 -> STM m [(ticketNo, objectId, m object)]
+  -- ^ Get the list of objects available in the pool with a ticketNo greater than the specified one. The number of returned objects is capped by the given Word64.
+  -- Only the IDs and ticketNos of the objects are directly accessible; each actual object must be loaded through a monadic action.
+  -- TODO: This signature assume that we have all the IDs and ticketNos in memory, but not the actual objects. This might change if IDs must be loaded from disk too.
   }
 
 data ObjectPoolWriter objectId object m
   = ObjectPoolWriter
-  { wrGetObjectId :: object -> objectId
+  { opwObjectId :: object -> objectId
   -- ^ Return the id of the specified object
-  , objectPoolAddObjects :: [object] -> m ()
+  , opwAddObjects :: [object] -> m ()
   -- ^ Add a batch of objects to the objectPool.
+  , opwHasObject :: m (objectId -> Bool)
+  -- ^ Check if the object pool contains an object with the given id
   }

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/ObjectPool/API.hs

@@ -5,7 +5,6 @@ module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.PerasCert
   , makePerasCertPoolWriterFromChainDB
   ) where
 
-import Data.Functor ((<&>))
 import Ouroboros.Consensus.Block
 import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
 import Ouroboros.Consensus.Storage.ChainDB.API (ChainDB)
@@ -24,18 +23,15 @@ makePerasCertPoolReader ::
   ObjectPoolReader PerasRoundNo (PerasCert blk) PerasCertTicketNo m
 makePerasCertPoolReader getCertSnapshot =
   ObjectPoolReader
-    { rdrGetObjectId = perasCertRound
-    , objectPoolGetSnapshot =
-        getCertSnapshot <&> \snap ->
-          ObjectPoolSnapshot
-            { objectPoolObjectsAfter = \ticketNo ->
-                [ (cert, tno, sz)
-                | (cert, tno) <- PerasCertDB.getCertsAfter snap ticketNo
-                , let sz = 0 -- TODO
-                ]
-            , objectPoolHasObject = PerasCertDB.containsCert snap
-            }
-    , objectPoolZeroTicketNo = PerasCertDB.zeroPerasCertTicketNo
+    { oprObjectId = perasCertRound
+    , oprZeroTicketNo = PerasCertDB.zeroPerasCertTicketNo
+    , oprObjectsAfter = \lastKnown limit -> do
+        certSnapshot <- getCertSnapshot
+        pure $
+          take (fromIntegral limit) $
+            [ (ticketNo, perasCertRound cert, pure cert)
+            | (cert, ticketNo) <- PerasCertDB.getCertsAfter certSnapshot lastKnown
+            ]
     }
 
 makePerasCertPoolReaderFromCertDB ::
@@ -45,13 +41,16 @@ makePerasCertPoolReaderFromCertDB perasCertDB =
   makePerasCertPoolReader (PerasCertDB.getCertSnapshot perasCertDB)
 
 makePerasCertPoolWriterFromCertDB ::
-  (StandardHash blk, Monad m) =>
+  (StandardHash blk, MonadSTM m) =>
   PerasCertDB m blk -> ObjectPoolWriter PerasRoundNo (PerasCert blk) m
 makePerasCertPoolWriterFromCertDB perasCertDB =
   ObjectPoolWriter
-    { wrGetObjectId = perasCertRound
-    , objectPoolAddObjects =
+    { opwObjectId = perasCertRound
+    , opwAddObjects =
         mapM_ $ PerasCertDB.addCert perasCertDB
+    , opwHasObject = do
+        certSnapshot <- atomically $ PerasCertDB.getCertSnapshot perasCertDB
+        pure $ PerasCertDB.containsCert certSnapshot
     }
 
 makePerasCertPoolReaderFromChainDB ::
@@ -61,11 +60,14 @@ makePerasCertPoolReaderFromChainDB chainDB =
   makePerasCertPoolReader (ChainDB.getPerasCertSnapshot chainDB)
 
 makePerasCertPoolWriterFromChainDB ::
-  (StandardHash blk, Monad m) =>
+  (StandardHash blk, MonadSTM m) =>
   ChainDB m blk -> ObjectPoolWriter PerasRoundNo (PerasCert blk) m
 makePerasCertPoolWriterFromChainDB chainDB =
   ObjectPoolWriter
-    { wrGetObjectId = perasCertRound
-    , objectPoolAddObjects =
+    { opwObjectId = perasCertRound
+    , opwAddObjects =
         mapM_ $ ChainDB.addPerasCertAsync chainDB
+    , opwHasObject = do
+        certSnapshot <- atomically $ ChainDB.getPerasCertSnapshot chainDB
+        pure $ PerasCertDB.containsCert certSnapshot
     }