Wrap validated Peras certificates with arrival time

This commit wraps the existing ValidatedPerasCerts stored in the
PerasCertDB with their corresponding arrival time. In addition, it
adapts tests to use either a randomly generated arrival time, or (when
appropriate) one generated by a monotonically increasing SystemTime.

Co-authored-by: Agustin Mista <[email protected]>
Co-authored-by: Alexander Esgen <[email protected]>
Co-authored-by: Georgy Lukyanov <[email protected]>
Co-authored-by: Thomas BAGREL <[email protected]>
Co-authored-by: Nicolas BACQUEY <[email protected]>

Author: 5 people

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

@@ -263,6 +263,7 @@ mkHandlers
     , keepAliveRng
     , miniProtocolParameters
     , getDiffusionPipeliningSupport
+    , systemTime
     }
   NodeKernel
     { getChainDB
@@ -322,7 +323,7 @@ mkHandlers
             , 10 -- TODO: see https://github.com/tweag/cardano-peras/issues/97
             , 10 -- TODO: see https://github.com/tweag/cardano-peras/issues/97
             )
-            (makePerasCertPoolWriterFromChainDB $ getChainDB)
+            (makePerasCertPoolWriterFromChainDB systemTime getChainDB)
             version
             controlMessageSTM
       , hPerasCertDiffusionServer = \version peer ->

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

@@ -585,6 +585,7 @@ runWith RunNodeArgs{..} encAddrNtN decAddrNtN LowLevelRunNodeArgs{..} =
                   llrnFeatureFlags
                   rnTraceConsensus
                   btime
+                  systemTime
                   (InFutureCheck.realHeaderInFutureCheck llrnMaxClockSkew systemTime)
                   historicityCheck
                   chainDB
@@ -863,6 +864,7 @@ mkNodeKernelArgs ::
   Set CardanoFeatureFlag ->
   Tracers m (ConnectionId addrNTN) (ConnectionId addrNTC) blk ->
   BlockchainTime m ->
+  SystemTime m ->
   InFutureCheck.SomeHeaderInFutureCheck m blk ->
   (m GSM.GsmState -> HistoricityCheck m blk) ->
   ChainDB m blk ->
@@ -883,6 +885,7 @@ mkNodeKernelArgs
   featureFlags
   tracers
   btime
+  systemTime
   chainSyncFutureCheck
   chainSyncHistoricityCheck
   chainDB
@@ -902,6 +905,7 @@ mkNodeKernelArgs
           , cfg
           , featureFlags
           , btime
+          , systemTime
           , chainDB
           , initChainDB = nodeInitChainDB
           , chainSyncFutureCheck

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs

@@ -199,6 +199,7 @@ data NodeKernelArgs m addrNTN addrNTC blk = NodeKernelArgs
   , cfg :: TopLevelConfig blk
   , featureFlags :: Set CardanoFeatureFlag
   , btime :: BlockchainTime m
+  , systemTime :: SystemTime m
   , chainDB :: ChainDB m blk
   , initChainDB :: StorageConfig blk -> InitChainDB m blk -> m ()
   , chainSyncFutureCheck :: SomeHeaderInFutureCheck m blk

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

@@ -1047,6 +1047,7 @@ runThreadNetwork
               , cfg = pInfoConfig
               , featureFlags = mempty
               , btime
+              , systemTime
               , chainDB
               , initChainDB = nodeInitChainDB
               , chainSyncFutureCheck =

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

@@ -15,6 +15,11 @@ import Data.Map (Map)
 import qualified Data.Map as Map
 import GHC.Exception (throw)
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types
+  ( SystemTime (..)
+  , WithArrivalTime (..)
+  , addArrivalTime
+  )
 import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
 import Ouroboros.Consensus.Storage.ChainDB.API (ChainDB)
 import qualified Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
@@ -43,7 +48,10 @@ makePerasCertPoolReaderFromSnapshot getCertSnapshot =
         let certsAfterLastKnown =
               PerasCertDB.getCertsAfter certSnapshot lastKnown
         let loadCertsAfterLastKnown =
-              pure (vpcCert <$> takeAscMap (fromIntegral limit) certsAfterLastKnown)
+              pure $
+                fmap
+                  (vpcCert . forgetArrivalTime)
+                  (takeAscMap (fromIntegral limit) certsAfterLastKnown)
         pure $
           if Map.null certsAfterLastKnown
             then Nothing
@@ -58,13 +66,13 @@ makePerasCertPoolReaderFromCertDB perasCertDB =
 
 makePerasCertPoolWriterFromCertDB ::
   (StandardHash blk, IOLike m) =>
-  PerasCertDB m blk -> ObjectPoolWriter PerasRoundNo (PerasCert blk) m
-makePerasCertPoolWriterFromCertDB perasCertDB =
+  SystemTime m ->
+  PerasCertDB m blk ->
+  ObjectPoolWriter PerasRoundNo (PerasCert blk) m
+makePerasCertPoolWriterFromCertDB systemTime perasCertDB =
   ObjectPoolWriter
     { opwObjectId = getPerasCertRound
-    , opwAddObjects = \certs -> do
-        validatePerasCerts certs
-          >>= mapM_ (PerasCertDB.addCert perasCertDB)
+    , opwAddObjects = addPerasCerts systemTime (PerasCertDB.addCert perasCertDB)
     , opwHasObject = do
         certSnapshot <- PerasCertDB.getCertSnapshot perasCertDB
         pure $ PerasCertDB.containsCert certSnapshot
@@ -78,13 +86,13 @@ makePerasCertPoolReaderFromChainDB chainDB =
 
 makePerasCertPoolWriterFromChainDB ::
   (StandardHash blk, IOLike m) =>
-  ChainDB m blk -> ObjectPoolWriter PerasRoundNo (PerasCert blk) m
-makePerasCertPoolWriterFromChainDB chainDB =
+  SystemTime m ->
+  ChainDB m blk ->
+  ObjectPoolWriter PerasRoundNo (PerasCert blk) m
+makePerasCertPoolWriterFromChainDB systemTime chainDB =
   ObjectPoolWriter
     { opwObjectId = getPerasCertRound
-    , opwAddObjects = \certs -> do
-        validatePerasCerts certs
-          >>= mapM_ (ChainDB.addPerasCertAsync chainDB)
+    , opwAddObjects = addPerasCerts systemTime (ChainDB.addPerasCertAsync chainDB)
     , opwHasObject = do
         certSnapshot <- ChainDB.getPerasCertSnapshot chainDB
         pure $ PerasCertDB.containsCert certSnapshot
@@ -111,3 +119,23 @@ validatePerasCerts certs = do
   case traverse (validatePerasCert perasParams) certs of
     Left validationErr -> throw (PerasCertValidationError validationErr)
     Right validatedCerts -> return validatedCerts
+
+-- | Add a list of 'PerasCert's into an object pool.
+--
+-- NOTE: we first validate the certificates, throwing an exception if any of
+-- them are invalid. We then wrap them with their arrival time, and finally add
+-- them to the pool using the provided adder function.
+--
+-- The order of the first two operations (i.e., validation and timestamping) are
+-- rather arbitrary, and the abstract Peras protocol just assumes it can happen
+-- "within" a slot.
+addPerasCerts ::
+  (StandardHash blk, MonadThrow m) =>
+  SystemTime m ->
+  (WithArrivalTime (ValidatedPerasCert blk) -> m a) ->
+  [PerasCert blk] ->
+  m ()
+addPerasCerts systemTime adder certs = do
+  validatePerasCerts certs
+    >>= mapM (addArrivalTime systemTime)
+    >>= mapM_ adder

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/API.hs

@@ -81,6 +81,7 @@ import Control.ResourceRegistry
 import Data.Typeable (Typeable)
 import GHC.Generics (Generic)
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types (WithArrivalTime)
 import Ouroboros.Consensus.HeaderStateHistory
   ( HeaderStateHistory (..)
   )
@@ -396,7 +397,7 @@ data ChainDB m blk = ChainDB
   , getStatistics :: m (Maybe Statistics)
   -- ^ Get statistics from the LedgerDB, in particular the number of entries
   -- in the tables.
-  , addPerasCertAsync :: ValidatedPerasCert blk -> m (AddPerasCertPromise m)
+  , addPerasCertAsync :: WithArrivalTime (ValidatedPerasCert blk) -> m (AddPerasCertPromise m)
   -- ^ Asynchronously insert a certificate to the DB. If this leads to a fork to
   -- be weightier than our current selection, this will trigger a fork switch.
   , getPerasWeightSnapshot :: STM m (WithFingerprint (PerasWeightSnapshot blk))
@@ -537,7 +538,7 @@ newtype AddPerasCertPromise m = AddPerasCertPromise
   -- impossible).
   }
 
-addPerasCertSync :: IOLike m => ChainDB m blk -> ValidatedPerasCert blk -> m ()
+addPerasCertSync :: IOLike m => ChainDB m blk -> WithArrivalTime (ValidatedPerasCert blk) -> m ()
 addPerasCertSync chainDB cert =
   waitPerasCertProcessed =<< addPerasCertAsync chainDB cert
 

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/ChainSel.hs

@@ -45,6 +45,7 @@ import qualified Data.Set as Set
 import Data.Traversable (for)
 import GHC.Stack (HasCallStack)
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types (WithArrivalTime)
 import Ouroboros.Consensus.Config
 import Ouroboros.Consensus.Fragment.Diff (ChainDiff (..))
 import qualified Ouroboros.Consensus.Fragment.Diff as Diff
@@ -328,7 +329,7 @@ addPerasCertAsync ::
   forall m blk.
   IOLike m =>
   ChainDbEnv m blk ->
-  ValidatedPerasCert blk ->
+  WithArrivalTime (ValidatedPerasCert blk) ->
   m (AddPerasCertPromise m)
 addPerasCertAsync CDB{cdbTracer, cdbChainSelQueue} =
   addPerasCertToQueue (TraceAddPerasCertEvent >$< cdbTracer) cdbChainSelQueue

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Types.hs

@@ -94,6 +94,7 @@ import Data.Word (Word64)
 import GHC.Generics (Generic)
 import NoThunks.Class (OnlyCheckWhnfNamed (..))
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types (WithArrivalTime)
 import Ouroboros.Consensus.Config
 import Ouroboros.Consensus.Fragment.Diff (ChainDiff)
 import Ouroboros.Consensus.HeaderValidation (HeaderWithTime (..))
@@ -553,7 +554,7 @@ data ChainSelMessage m blk
     ChainSelAddBlock !(BlockToAdd m blk)
   | -- | Add a Peras certificate
     ChainSelAddPerasCert
-      !(ValidatedPerasCert blk)
+      !(WithArrivalTime (ValidatedPerasCert blk))
       -- | Used for 'AddPerasCertPromise'.
       !(StrictTMVar m ())
   | -- | Reprocess blocks that have been postponed by the LoE.
@@ -609,7 +610,7 @@ addPerasCertToQueue ::
   IOLike m =>
   Tracer m (TraceAddPerasCertEvent blk) ->
   ChainSelQueue m blk ->
-  ValidatedPerasCert blk ->
+  WithArrivalTime (ValidatedPerasCert blk) ->
   m (AddPerasCertPromise m)
 addPerasCertToQueue tracer ChainSelQueue{varChainSelQueue} cert = do
   varProcessed <- newEmptyTMVarIO

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/PerasCertDB/API.hs

@@ -17,12 +17,13 @@ import Data.Map (Map)
 import Data.Word (Word64)
 import NoThunks.Class
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types (WithArrivalTime)
 import Ouroboros.Consensus.Peras.Weight
 import Ouroboros.Consensus.Util.IOLike
 import Ouroboros.Consensus.Util.STM (WithFingerprint (..))
 
 data PerasCertDB m blk = PerasCertDB
-  { addCert :: ValidatedPerasCert blk -> m AddPerasCertResult
+  { addCert :: WithArrivalTime (ValidatedPerasCert blk) -> m AddPerasCertResult
   -- ^ Add a Peras certificate to the database. The result indicates whether
   -- the certificate was actually added, or if it was already present.
   , getWeightSnapshot :: STM m (WithFingerprint (PerasWeightSnapshot blk))
@@ -46,7 +47,9 @@ data AddPerasCertResult = AddedPerasCertToDB | PerasCertAlreadyInDB
 data PerasCertSnapshot blk = PerasCertSnapshot
   { containsCert :: PerasRoundNo -> Bool
   -- ^ Do we have the certificate for this round?
-  , getCertsAfter :: PerasCertTicketNo -> Map PerasCertTicketNo (ValidatedPerasCert blk)
+  , getCertsAfter ::
+      PerasCertTicketNo ->
+      Map PerasCertTicketNo (WithArrivalTime (ValidatedPerasCert blk))
   -- ^ Get certificates after the given ticket number (excluded).
   -- The result is a map of ticket numbers to validated certificates.
   }

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/PerasCertDB/Impl.hs

@@ -32,6 +32,7 @@ import qualified Data.Set as Set
 import GHC.Generics (Generic)
 import NoThunks.Class
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.BlockchainTime.WallClock.Types (WithArrivalTime)
 import Ouroboros.Consensus.Peras.Weight
 import Ouroboros.Consensus.Storage.PerasCertDB.API
 import Ouroboros.Consensus.Util.Args
@@ -152,7 +153,7 @@ implAddCert ::
   , StandardHash blk
   ) =>
   PerasCertDbEnv m blk ->
-  ValidatedPerasCert blk ->
+  WithArrivalTime (ValidatedPerasCert blk) ->
   m AddPerasCertResult
 implAddCert env cert = do
   traceWith pcdbTracer $ AddingPerasCert roundNo boostedPt
@@ -255,11 +256,13 @@ implGarbageCollect PerasCertDbEnv{pcdbVolatileState} slot =
 --
 -- INVARIANT: See 'invariantForPerasVolatileCertState'.
 data PerasVolatileCertState blk = PerasVolatileCertState
-  { pvcsCerts :: !(Map PerasRoundNo (ValidatedPerasCert blk))
+  { pvcsCerts :: !(Map PerasRoundNo (WithArrivalTime (ValidatedPerasCert blk)))
   -- ^ The boosted blocks by 'RoundNo' of all certificates currently in the db.
   , pvcsWeightByPoint :: !(PerasWeightSnapshot blk)
   -- ^ The weight of boosted blocks w.r.t. the certificates currently in the db.
-  , pvcsCertsByTicket :: !(Map PerasCertTicketNo (ValidatedPerasCert blk))
+  --
+  -- INVARIANT: In sync with 'pvcsCerts'.
+  , pvcsCertsByTicket :: !(Map PerasCertTicketNo (WithArrivalTime (ValidatedPerasCert blk)))
   -- ^ The certificates by 'PerasCertTicketNo'.
   , pvcsLastTicketNo :: !PerasCertTicketNo
   -- ^ The most recent 'PerasCertTicketNo' (or 'zeroPerasCertTicketNo'