ChainDB: implement chain selection for certificates

Author: amesgen

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

@@ -25,6 +25,10 @@ module Ouroboros.Consensus.Storage.ChainDB.API
   , addBlockWaitWrittenToDisk
   , addBlock_
 
+    -- * Adding a Peras certificate
+  , AddPerasCertPromise (..)
+  , addPerasCertSync
+
     -- * Trigger chain selection
   , ChainSelectionPromise (..)
   , triggerChainSelection
@@ -387,7 +391,7 @@ data ChainDB m blk = ChainDB
   , getStatistics :: m (Maybe Statistics)
   -- ^ Get statistics from the LedgerDB, in particular the number of entries
   -- in the tables.
-  , addPerasCert :: PerasCert blk -> m ()
+  , addPerasCertAsync :: PerasCert blk -> m (AddPerasCertPromise m)
   -- ^ TODO
   , getPerasWeightSnapshot :: STM m (WithFingerprint (PerasWeightSnapshot blk))
   -- ^ TODO
@@ -510,6 +514,23 @@ triggerChainSelection :: IOLike m => ChainDB m blk -> m ()
 triggerChainSelection chainDB =
   waitChainSelectionPromise =<< chainSelAsync chainDB
 
+{-------------------------------------------------------------------------------
+  Adding a Peras certificate
+-------------------------------------------------------------------------------}
+
+newtype AddPerasCertPromise m = AddPerasCertPromise
+  { waitPerasCertProcessed :: m ()
+  -- ^ Wait until the Peras certificate has been processed (which potentially
+  -- includes switching to a different chain). If the PerasCertDB did already
+  -- contain a certificate for this round, the certificate is ignored (as the
+  -- two certificates must be identical because certificate equivocation is
+  -- impossible).
+  }
+
+addPerasCertSync :: IOLike m => ChainDB m blk -> PerasCert blk -> m ()
+addPerasCertSync chainDB cert =
+  waitPerasCertProcessed =<< addPerasCertAsync chainDB cert
+
 {-------------------------------------------------------------------------------
   Serialised block/header with its point
 -------------------------------------------------------------------------------}

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

@@ -16,6 +16,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl
     -- * Trace types
   , SelectionChangedInfo (..)
   , TraceAddBlockEvent (..)
+  , TraceAddPerasCertEvent (..)
   , TraceChainSelStarvationEvent (..)
   , TraceCopyToImmutableDBEvent (..)
   , TraceEvent (..)
@@ -278,10 +279,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
             , getHeaderStateHistory = getEnvSTM h Query.getHeaderStateHistory
             , getReadOnlyForkerAtPoint = getEnv2 h Query.getReadOnlyForkerAtPoint
             , getStatistics = getEnv h Query.getStatistics
-            , addPerasCert = getEnv1 h $ \cdb@CDB{..} cert -> do
-                _ <- PerasCertDB.addCert cdbPerasCertDB cert
-                -- TODO trigger chain selection in a more efficient way
-                waitChainSelectionPromise =<< ChainSel.triggerChainSelectionAsync cdb
+            , addPerasCertAsync = getEnv1 h ChainSel.addPerasCertAsync
             , getPerasWeightSnapshot = getEnvSTM h Query.getPerasWeightSnapshot
             }
     addBlockTestFuse <- newFuse "test chain selection"

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

@@ -546,6 +546,8 @@ addBlockRunner fuse cdb@CDB{..} = forever $ do
                   varBlockProcessed
                   (FailedToAddBlock "Failed to add block synchronously")
               pure ()
+            ChainSelAddPerasCert _cert varProcessed ->
+              void $ tryPutTMVar varProcessed ()
           closeChainSelQueue cdbChainSelQueue
       )
       ( \message -> do
@@ -554,6 +556,10 @@ addBlockRunner fuse cdb@CDB{..} = forever $ do
               trace PoppedReprocessLoEBlocksFromQueue
             ChainSelAddBlock BlockToAdd{blockToAdd} ->
               trace $ PoppedBlockFromQueue $ blockRealPoint blockToAdd
+            ChainSelAddPerasCert cert _varProcessed ->
+              traceWith cdbTracer $
+                TraceAddPerasCertEvent $
+                  PoppedPerasCertFromQueue (perasCertRound cert) (perasCertBoostedBlock cert)
           chainSelSync cdb message
           lift $ atomically $ processedChainSelMessage cdbChainSelQueue message
       )

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

@@ -13,6 +13,7 @@
 -- adding a block.
 module Ouroboros.Consensus.Storage.ChainDB.Impl.ChainSel
   ( addBlockAsync
+  , addPerasCertAsync
   , chainSelSync
   , chainSelectionForBlock
   , initialChainSelection
@@ -68,6 +69,7 @@ import Ouroboros.Consensus.Peras.Weight
 import Ouroboros.Consensus.Storage.ChainDB.API
   ( AddBlockPromise (..)
   , AddBlockResult (..)
+  , AddPerasCertPromise
   , BlockComponent (..)
   , ChainType (..)
   , LoE (..)
@@ -91,10 +93,12 @@ import Ouroboros.Consensus.Storage.ImmutableDB (ImmutableDB)
 import qualified Ouroboros.Consensus.Storage.ImmutableDB as ImmutableDB
 import Ouroboros.Consensus.Storage.LedgerDB
 import qualified Ouroboros.Consensus.Storage.LedgerDB as LedgerDB
+import qualified Ouroboros.Consensus.Storage.PerasCertDB.API as PerasCertDB
 import Ouroboros.Consensus.Storage.VolatileDB (VolatileDB)
 import qualified Ouroboros.Consensus.Storage.VolatileDB as VolatileDB
 import Ouroboros.Consensus.Util
 import Ouroboros.Consensus.Util.AnchoredFragment
+import Ouroboros.Consensus.Util.EarlyExit (exitEarly, withEarlyExit_)
 import Ouroboros.Consensus.Util.Enclose (encloseWith)
 import Ouroboros.Consensus.Util.IOLike
 import Ouroboros.Consensus.Util.STM (WithFingerprint (..))
@@ -323,6 +327,15 @@ addBlockAsync ::
 addBlockAsync CDB{cdbTracer, cdbChainSelQueue} =
   addBlockToAdd (TraceAddBlockEvent >$< cdbTracer) cdbChainSelQueue
 
+addPerasCertAsync ::
+  forall m blk.
+  (IOLike m, HasHeader blk) =>
+  ChainDbEnv m blk ->
+  PerasCert blk ->
+  m (AddPerasCertPromise m)
+addPerasCertAsync CDB{cdbTracer, cdbChainSelQueue} =
+  addPerasCertToQueue (TraceAddPerasCertEvent >$< cdbTracer) cdbChainSelQueue
+
 -- | Schedule reprocessing of blocks postponed by the LoE.
 triggerChainSelectionAsync ::
   forall m blk.
@@ -461,6 +474,65 @@ chainSelSync cdb@CDB{..} (ChainSelAddBlock BlockToAdd{blockToAdd = b, ..}) = do
   deliverProcessed tip =
     atomically $
       putTMVar varBlockProcessed (SuccesfullyAddedBlock tip)
+-- Process a Peras certificate by adding it to the PerasCertDB and potentially
+-- performing chain selection if a candidate is now better than our selection.
+chainSelSync cdb@CDB{..} (ChainSelAddPerasCert cert varProcessed) = do
+  curChain <- lift $ atomically $ Query.getCurrentChain cdb
+  let immTip = castPoint $ AF.anchorPoint curChain
+
+  withEarlyExit_ $ do
+    -- Ignore the certificate if it boosts a block that is so old that it can't
+    -- influence our selection.
+    when (pointSlot boostedBlock < pointSlot immTip) $ do
+      lift $ lift $ traceWith tracer $ IgnorePerasCertTooOld certRound boostedBlock immTip
+      exitEarly
+
+    -- Add the certificate to the PerasCertDB.
+    lift (lift $ PerasCertDB.addCert cdbPerasCertDB cert) >>= \case
+      PerasCertDB.AddedPerasCertToDB -> pure ()
+      -- If it already is in the PerasCertDB, we are done.
+      PerasCertDB.PerasCertAlreadyInDB -> exitEarly
+
+    -- If the certificate boosts a block on our current chain (including the
+    -- anchor), then it just makes our selection even stronger.
+    when (AF.withinFragmentBounds (castPoint boostedBlock) curChain) $ do
+      lift $ lift $ traceWith tracer $ PerasCertBoostsCurrentChain certRound boostedBlock
+      exitEarly
+
+    boostedHash <- case pointHash boostedBlock of
+      -- If the certificate boosts the Genesis point, then it can not influence
+      -- chain selection as all chains contain it.
+      GenesisHash -> do
+        lift $ lift $ traceWith tracer $ PerasCertBoostsGenesis certRound
+        exitEarly
+      -- Otherwise, the certificate boosts a block potentially on a (future)
+      -- candidate.
+      BlockHash boostedHash -> pure boostedHash
+    boostedHdr <-
+      lift (lift $ VolatileDB.getBlockComponent cdbVolatileDB GetHeader boostedHash) >>= \case
+        -- If we have not (yet) received the boosted block, we don't need to do
+        -- anything further for now regarding chain selection. Once we receive
+        -- it, the additional weight of the certificate is taken into account.
+        Nothing -> do
+          lift $ lift $ traceWith tracer $ PerasCertBoostsBlockNotYetReceived certRound boostedBlock
+          exitEarly
+        Just boostedHdr -> pure boostedHdr
+
+    -- Trigger chain selection for the boosted block.
+    lift $ lift $ traceWith tracer $ ChainSelectionForBoostedBlock certRound boostedBlock
+    lift $ chainSelectionForBlock cdb BlockCache.empty boostedHdr noPunishment
+
+  -- Deliver promise indicating that we processed the cert.
+  lift $ atomically $ putTMVar varProcessed ()
+ where
+  tracer :: Tracer m (TraceAddPerasCertEvent blk)
+  tracer = TraceAddPerasCertEvent >$< cdbTracer
+
+  certRound :: PerasRoundNo
+  certRound = perasCertRound cert
+
+  boostedBlock :: Point blk
+  boostedBlock = perasCertBoostedBlock cert
 
 -- | Return 'True' when the given header should be ignored when adding it
 -- because it is too old, i.e., we wouldn't be able to switch to a chain

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

@@ -55,6 +55,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types
   , ChainSelMessage (..)
   , ChainSelQueue -- opaque
   , addBlockToAdd
+  , addPerasCertToQueue
   , addReprocessLoEBlocks
   , closeChainSelQueue
   , getChainSelMessage
@@ -66,6 +67,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types
     -- * Trace types
   , SelectionChangedInfo (..)
   , TraceAddBlockEvent (..)
+  , TraceAddPerasCertEvent (..)
   , TraceChainSelStarvationEvent (..)
   , TraceCopyToImmutableDBEvent (..)
   , TraceEvent (..)
@@ -83,7 +85,6 @@ import Control.ResourceRegistry
 import Control.Tracer
 import Data.Foldable (traverse_)
 import Data.Map.Strict (Map)
-import Data.Maybe (mapMaybe)
 import Data.Maybe.Strict (StrictMaybe (..))
 import Data.MultiSet (MultiSet)
 import qualified Data.MultiSet as MultiSet
@@ -104,6 +105,7 @@ import Ouroboros.Consensus.Protocol.Abstract
 import Ouroboros.Consensus.Storage.ChainDB.API
   ( AddBlockPromise (..)
   , AddBlockResult (..)
+  , AddPerasCertPromise (..)
   , ChainDbError (..)
   , ChainSelectionPromise (..)
   , ChainType
@@ -549,6 +551,11 @@ data BlockToAdd m blk = BlockToAdd
 data ChainSelMessage m blk
   = -- | Add a new block
     ChainSelAddBlock !(BlockToAdd m blk)
+  | -- | Add a Peras certificate
+    ChainSelAddPerasCert
+      !(PerasCert blk)
+      -- | Used for 'AddPerasCertPromise'.
+      !(StrictTMVar m ())
   | -- | Reprocess blocks that have been postponed by the LoE.
     ChainSelReprocessLoEBlocks
       -- | Used for 'ChainSelectionPromise'.
@@ -597,6 +604,28 @@ addBlockToAdd tracer (ChainSelQueue{varChainSelQueue, varChainSelPoints}) punish
       , blockProcessed = readTMVar varBlockProcessed
       }
 
+-- | Add a Peras certificate to the background queue.
+addPerasCertToQueue ::
+  (IOLike m, StandardHash blk) =>
+  Tracer m (TraceAddPerasCertEvent blk) ->
+  ChainSelQueue m blk ->
+  PerasCert blk ->
+  m (AddPerasCertPromise m)
+addPerasCertToQueue tracer ChainSelQueue{varChainSelQueue} cert = do
+  varProcessed <- newEmptyTMVarIO
+  traceWith tracer $ addedToQueue RisingEdge
+  queueSize <- atomically $ do
+    writeTBQueue varChainSelQueue $ ChainSelAddPerasCert cert varProcessed
+    lengthTBQueue varChainSelQueue
+  traceWith tracer $ addedToQueue $ FallingEdgeWith $ fromIntegral queueSize
+  pure
+    AddPerasCertPromise
+      { waitPerasCertProcessed = atomically $ takeTMVar varProcessed
+      }
+ where
+  addedToQueue =
+    AddedPerasCertToQueue (perasCertRound cert) (perasCertBoostedBlock cert)
+
 -- | Try to add blocks again that were postponed due to the LoE.
 addReprocessLoEBlocks ::
   IOLike m =>
@@ -651,6 +680,7 @@ getChainSelMessage starvationTracer starvationVar chainSelQueue =
       let pt = blockRealPoint block
       traceWith starvationTracer $ ChainSelStarvation (FallingEdgeWith pt)
       atomically . writeTVar starvationVar . ChainSelStarvationEndedAt =<< getMonotonicTime
+    ChainSelAddPerasCert{} -> pure ()
     ChainSelReprocessLoEBlocks{} -> pure ()
 
 -- TODO Can't use tryReadTBQueue from io-classes because it is broken for IOSim
@@ -661,18 +691,15 @@ tryReadTBQueue' q = (Just <$> readTBQueue q) `orElse` pure Nothing
 -- | Flush the 'ChainSelQueue' queue and notify the waiting threads.
 closeChainSelQueue :: IOLike m => ChainSelQueue m blk -> STM m ()
 closeChainSelQueue ChainSelQueue{varChainSelQueue = queue} = do
-  as <- mapMaybe blockAdd <$> flushTBQueue queue
-  traverse_
-    ( \a ->
-        tryPutTMVar
-          (varBlockProcessed a)
-          (FailedToAddBlock "Queue flushed")
-    )
-    as
+  traverse_ deliverPromise =<< flushTBQueue queue
  where
-  blockAdd = \case
-    ChainSelAddBlock ab -> Just ab
-    ChainSelReprocessLoEBlocks _ -> Nothing
+  deliverPromise = \case
+    ChainSelAddBlock ab ->
+      tryPutTMVar (varBlockProcessed ab) (FailedToAddBlock "Queue flushed")
+    ChainSelAddPerasCert _cert varProcessed ->
+      tryPutTMVar varProcessed ()
+    ChainSelReprocessLoEBlocks varProcessed ->
+      tryPutTMVar varProcessed ()
 
 -- | To invoke when the given 'ChainSelMessage' has been processed by ChainSel.
 -- This is used to remove the respective point from the multiset of points in
@@ -685,6 +712,8 @@ processedChainSelMessage ::
 processedChainSelMessage ChainSelQueue{varChainSelPoints} = \case
   ChainSelAddBlock BlockToAdd{blockToAdd = blk} ->
     modifyTVar varChainSelPoints $ MultiSet.delete (blockRealPoint blk)
+  ChainSelAddPerasCert{} ->
+    pure ()
   ChainSelReprocessLoEBlocks{} ->
     pure ()
 
@@ -729,6 +758,7 @@ data TraceEvent blk
   | TracePerasCertDbEvent (PerasCertDB.TraceEvent blk)
   | TraceLastShutdownUnclean
   | TraceChainSelStarvationEvent (TraceChainSelStarvationEvent blk)
+  | TraceAddPerasCertEvent (TraceAddPerasCertEvent blk)
   deriving Generic
 
 deriving instance
@@ -1035,3 +1065,26 @@ data TraceIteratorEvent blk
 newtype TraceChainSelStarvationEvent blk
   = ChainSelStarvation (Enclosing' (RealPoint blk))
   deriving (Generic, Eq, Show)
+
+data TraceAddPerasCertEvent blk
+  = -- | The Peras certificate from the given round boosting the given block was
+    -- added to the queue. The size of the queue is included.
+    AddedPerasCertToQueue PerasRoundNo (Point blk) (Enclosing' Word)
+  | -- | The Peras certificate from the given round boosting the given block was
+    -- popped from the queue.
+    PoppedPerasCertFromQueue PerasRoundNo (Point blk)
+  | -- | The Peras certificate from the given round boosting the given block was
+    -- too old, ie its slot was older than the current immutable slot (the third
+    -- argument).
+    IgnorePerasCertTooOld PerasRoundNo (Point blk) (Point blk)
+  | -- | The Peras certificate from the given round boosts a block on the
+    -- current selection.
+    PerasCertBoostsCurrentChain PerasRoundNo (Point blk)
+  | -- | The Peras certificate from the given round boosts the Genesis point.
+    PerasCertBoostsGenesis PerasRoundNo
+  | -- | The Peras certificate from the given round boosts a block that we have
+    -- not (yet) received.
+    PerasCertBoostsBlockNotYetReceived PerasRoundNo (Point blk)
+  | -- | Perform chain selection for a block boosted by a Peras certificate.
+    ChainSelectionForBoostedBlock PerasRoundNo (Point blk)
+  deriving (Generic, Eq, Show)

ouroboros-consensus/test/storage-test/Test/Ouroboros/Storage/ChainDB/StateMachine.hs

@@ -1352,6 +1352,8 @@ deriving instance SOP.Generic (PerasCertDB.TraceEvent blk)
 deriving instance SOP.HasDatatypeInfo (PerasCertDB.TraceEvent blk)
 deriving anyclass instance SOP.Generic (TraceChainSelStarvationEvent blk)
 deriving anyclass instance SOP.HasDatatypeInfo (TraceChainSelStarvationEvent blk)
+deriving anyclass instance SOP.Generic (TraceAddPerasCertEvent blk)
+deriving anyclass instance SOP.HasDatatypeInfo (TraceAddPerasCertEvent blk)
 
 data Tag
   = TagGetIsValidJust
@@ -1779,6 +1781,7 @@ traceEventName = \case
   TracePerasCertDbEvent ev -> "PerasCertDB." <> constrName ev
   TraceLastShutdownUnclean -> "LastShutdownUnclean"
   TraceChainSelStarvationEvent ev -> "ChainSelStarvation." <> constrName ev
+  TraceAddPerasCertEvent ev -> "AddPerasCert." <> constrName ev
 
 mkArgs ::
   IOLike m =>