ChainDB: pass new block to ledger validation

In `ChainDB.addBlock`, we receive a new block and use it to try switch to a
longer chain. This will require applying the block to the ledger, which
requires reading the block from disk and parsing it again. This is right in
the critical path of (bulk) chain sync.

Since we have the block in memory, we can avoid the redundant read and pass it
to the validation code directly in the form of a `BlockCache`. We can later
use this cache for caching more/other blocks.

Note that when switching to a fork or when we can extend the new chain with
more blocks after the new block, we'll still have to read blocks from disk in
order to validate them, but not the new block.

ouroboros-consensus/ouroboros-consensus.cabal

@@ -145,6 +145,7 @@ library
                        Ouroboros.Storage.ChainDB.Impl
                        Ouroboros.Storage.ChainDB.Impl.Args
                        Ouroboros.Storage.ChainDB.Impl.Background
+                       Ouroboros.Storage.ChainDB.Impl.BlockCache
                        Ouroboros.Storage.ChainDB.Impl.BlockComponent
                        Ouroboros.Storage.ChainDB.Impl.ChainSel
                        Ouroboros.Storage.ChainDB.Impl.ImmDB

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

@@ -58,6 +58,7 @@ import           Ouroboros.Consensus.Util (whenJust)
 import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Consensus.Util.ResourceRegistry
 
+import qualified Ouroboros.Storage.ChainDB.Impl.BlockCache as BlockCache
 import           Ouroboros.Storage.ChainDB.Impl.ChainSel
                      (chainSelectionForBlock)
 import qualified Ouroboros.Storage.ChainDB.Impl.ImmDB as ImmDB
@@ -405,7 +406,7 @@ scheduledChainSelection cdb@CDB{..} curSlot = do
       -- which case, the ChainDB has to be (re)started, triggering a full
       -- chain selection, which would include these blocks. So there is no
       -- risk of "forgetting" to add a block.
-      mapM_ (chainSelectionForBlock cdb) hdrs
+      mapM_ (chainSelectionForBlock cdb BlockCache.empty) hdrs
 
 -- | Whenever the current slot changes, call 'scheduledChainSelection' for the
 -- (new) current slot.

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

@@ -0,0 +1,48 @@
+{-# LANGUAGE FlexibleContexts #-}
+-- | Cache blocks in memory
+--
+-- Intended for qualified import.
+--
+-- > import           Ouroboros.Storage.ChainDB.Impl.BlockCache (BlockCache)
+-- > import qualified Ouroboros.Storage.ChainDB.Impl.BlockCache as BlockCache
+module Ouroboros.Storage.ChainDB.Impl.BlockCache
+  ( BlockCache -- opaque
+  , empty
+  , singleton
+  , cacheBlock
+  , lookup
+  , toHeaderOrBlock
+  ) where
+
+import           Prelude hiding (lookup)
+
+import           Data.Map (Map)
+import qualified Data.Map as Map
+
+import           Ouroboros.Network.Block (HasHeader (..), HeaderHash)
+
+import           Ouroboros.Consensus.Block (Header, headerHash)
+
+
+newtype BlockCache blk = BlockCache (Map (HeaderHash blk) blk)
+
+empty :: BlockCache blk
+empty = BlockCache Map.empty
+
+singleton :: HasHeader blk => blk -> BlockCache blk
+singleton blk = cacheBlock blk empty
+
+cacheBlock :: HasHeader blk => blk -> BlockCache blk -> BlockCache blk
+cacheBlock blk (BlockCache cache) = BlockCache (Map.insert (blockHash blk) blk cache)
+
+lookup :: HasHeader blk => HeaderHash blk -> BlockCache blk -> Maybe blk
+lookup hash (BlockCache cache) = Map.lookup hash cache
+
+toHeaderOrBlock
+  :: (HasHeader blk, HasHeader (Header blk))
+  => Header blk -> BlockCache blk -> Either (Header blk) blk
+toHeaderOrBlock hdr blockCache
+    | Just blk <- lookup (headerHash hdr) blockCache
+    = Right blk
+    | otherwise
+    = Left hdr

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

@@ -62,6 +62,8 @@ import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Consensus.Util.STM (WithFingerprint (..))
 
 import           Ouroboros.Storage.ChainDB.API (InvalidBlockReason (..))
+import           Ouroboros.Storage.ChainDB.Impl.BlockCache (BlockCache)
+import qualified Ouroboros.Storage.ChainDB.Impl.BlockCache as BlockCache
 import           Ouroboros.Storage.ChainDB.Impl.ImmDB (ImmDB)
 import qualified Ouroboros.Storage.ChainDB.Impl.ImmDB as ImmDB
 import           Ouroboros.Storage.ChainDB.Impl.LgrDB (LgrDB)
@@ -152,6 +154,7 @@ initialChainSelection immDB volDB lgrDB tracer cfg varInvalid curSlot = do
         (contramap (TraceInitChainSelEvent . InitChainSelValidation) tracer)
         cfg
         varInvalid
+        BlockCache.empty
         curChainAndLedger
         (fmap (mkCandidateSuffix 0) candidates)
 
@@ -205,7 +208,7 @@ addBlock cdb@CDB{..} b = do
       | otherwise -> do
         VolDB.putBlock cdbVolDB b
         trace $ AddedBlockToVolDB (blockPoint b) (blockNo b) (cdbIsEBB hdr)
-        chainSelectionForBlock cdb hdr
+        chainSelectionForBlock cdb (BlockCache.singleton b) hdr
   where
     trace :: TraceAddBlockEvent blk -> m ()
     trace = traceWith (contramap TraceAddBlockEvent cdbTracer)
@@ -266,9 +269,10 @@ chainSelectionForBlock
      , HasCallStack
      )
   => ChainDbEnv m blk
+  -> BlockCache blk
   -> Header blk
   -> m ()
-chainSelectionForBlock cdb@CDB{..} hdr = do
+chainSelectionForBlock cdb@CDB{..} blockCache hdr = do
     curSlot <- atomically $ getCurrentSlot cdbBlockchainTime
 
     (invalid, isMember, succsOf, predecessor, curChain, tipPoint, ledgerDB, immBlockNo)
@@ -446,6 +450,7 @@ chainSelectionForBlock cdb@CDB{..} hdr = do
       (contramap (TraceAddBlockEvent . AddBlockValidation) cdbTracer)
       cdbNodeConfig
       cdbInvalid
+      blockCache
 
     -- | Try to swap the current (chain) fragment with the given candidate
     -- fragment. The 'LgrDB.LedgerDB' is updated in the same transaction.
@@ -606,13 +611,14 @@ chainSelection
   -> Tracer m (TraceValidationEvent blk)
   -> NodeConfig (BlockProtocol blk)
   -> StrictTVar m (WithFingerprint (InvalidBlocks blk))
+  -> BlockCache blk
   -> ChainAndLedger blk              -- ^ The current chain and ledger
   -> NonEmpty (CandidateSuffix blk)  -- ^ Candidates
   -> m (Maybe (ChainAndLedger blk))
      -- ^ The (valid) chain and corresponding LedgerDB that was selected, or
      -- 'Nothing' if there is no valid chain preferred over the current
      -- chain.
-chainSelection lgrDB tracer cfg varInvalid
+chainSelection lgrDB tracer cfg varInvalid blockCache
                curChainAndLedger@(ChainAndLedger curChain _) candidates =
   assert (all (preferAnchoredCandidate cfg curChain . _suffix) candidates) $
   assert (all (isJust . fitCandidateSuffixOn curChain) candidates) $
@@ -625,7 +631,7 @@ chainSelection lgrDB tracer cfg varInvalid
     validate :: ChainAndLedger  blk  -- ^ Current chain and ledger
              -> CandidateSuffix blk  -- ^ Candidate fragment
              -> m (Maybe (ChainAndLedger blk))
-    validate = validateCandidate lgrDB tracer cfg varInvalid
+    validate = validateCandidate lgrDB tracer cfg varInvalid blockCache
 
     -- 1. Take the first candidate from the list of sorted candidates
     -- 2. Validate it
@@ -717,12 +723,13 @@ validateCandidate
   -> Tracer m (TraceValidationEvent blk)
   -> NodeConfig (BlockProtocol blk)
   -> StrictTVar m (WithFingerprint (InvalidBlocks blk))
+  -> BlockCache blk
   -> ChainAndLedger  blk                   -- ^ Current chain and ledger
   -> CandidateSuffix blk                   -- ^ Candidate fragment
   -> m (Maybe (ChainAndLedger blk))
-validateCandidate lgrDB tracer cfg varInvalid
+validateCandidate lgrDB tracer cfg varInvalid blockCache
                   (ChainAndLedger curChain curLedger) candSuffix =
-    LgrDB.validate lgrDB curLedger rollback newBlocks >>= \case
+    LgrDB.validate lgrDB curLedger blockCache rollback newBlocks >>= \case
       LgrDB.MaximumRollbackExceeded supported _ -> do
         trace $ CandidateExceedsRollback {
             _supportedRollback = supported

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

@@ -72,7 +72,7 @@ import           Control.Tracer
 
 import           Ouroboros.Network.Block (pattern BlockPoint,
                      pattern GenesisPoint, HasHeader (..), HeaderHash, Point,
-                     SlotNo, blockPoint, castPoint)
+                     SlotNo, blockPoint)
 import qualified Ouroboros.Network.Block as Block
 import           Ouroboros.Network.Point (WithOrigin (At))
 
@@ -104,6 +104,8 @@ import           Ouroboros.Storage.LedgerDB.OnDisk (DiskSnapshot,
 import qualified Ouroboros.Storage.LedgerDB.OnDisk as LedgerDB
 
 import           Ouroboros.Storage.ChainDB.API (ChainDbFailure (..))
+import           Ouroboros.Storage.ChainDB.Impl.BlockCache (BlockCache)
+import qualified Ouroboros.Storage.ChainDB.Impl.BlockCache as BlockCache
 import           Ouroboros.Storage.ChainDB.Impl.ImmDB (ImmDB)
 import qualified Ouroboros.Storage.ChainDB.Impl.ImmDB as ImmDB
 
@@ -382,10 +384,11 @@ validate :: forall m blk. (IOLike m, ProtocolLedgerView blk, HasCallStack)
          -> LedgerDB blk
             -- ^ This is used as the starting point for validation, not the one
             -- in the 'LgrDB'.
+         -> BlockCache blk
          -> Word64  -- ^ How many blocks to roll back
          -> [Header blk]
          -> m (ValidateResult blk)
-validate LgrDB{..} ledgerDB numRollbacks = \hdrs -> do
+validate LgrDB{..} ledgerDB blockCache numRollbacks = \hdrs -> do
     blocks <- toBlocks hdrs <$> atomically (readTVar varPrevApplied)
     res <- LedgerDB.ledgerDbSwitch conf numRollbacks blocks ledgerDB
     atomically $ modifyTVar varPrevApplied $
@@ -397,7 +400,7 @@ validate LgrDB{..} ledgerDB numRollbacks = \hdrs -> do
     toBlocks hdrs prevApplied =
       [ ( if Set.member (headerPoint hdr) prevApplied
           then Reapply else Apply
-        , toRefOrVal (Left hdr) )
+        , toRefOrVal $ BlockCache.toHeaderOrBlock hdr blockCache)
       | hdr <- hdrs ]
 
     -- | Based on the 'ValidateResult', return the hashes corresponding to
@@ -472,5 +475,5 @@ wrapFailure LgrDbArgs{ lgrHasFS = hasFS } k =
 
 toRefOrVal :: (HasHeader blk, HasHeader (Header blk))
            => Either (Header blk) blk -> RefOrVal (Point blk) blk
-toRefOrVal (Left  hdr) = Ref (castPoint (blockPoint hdr))
-toRefOrVal (Right blk) = Val (blockPoint blk) blk
+toRefOrVal (Left  hdr) = Ref (headerPoint hdr)
+toRefOrVal (Right blk) = Val (blockPoint  blk) blk

ouroboros-consensus/test-consensus/Test/Consensus/LocalStateQueryServer.hs

@@ -43,6 +43,7 @@ import           Ouroboros.Consensus.Protocol.BFT
 import           Ouroboros.Consensus.Util ((.:))
 import           Ouroboros.Consensus.Util.IOLike
 
+import qualified Ouroboros.Storage.ChainDB.Impl.BlockCache as BlockCache
 import qualified Ouroboros.Storage.ChainDB.Impl.LedgerCursor as LedgerCursor
 import           Ouroboros.Storage.ChainDB.Impl.LgrDB (LedgerDbParams (..),
                      LgrDB, LgrDbArgs (..), mkLgrDB)
@@ -181,7 +182,7 @@ initLgrDB k chain = do
     varDB          <- newTVarM genesisLedgerDB
     varPrevApplied <- newTVarM mempty
     let lgrDB = mkLgrDB conf varDB varPrevApplied args
-    LgrDB.validate lgrDB genesisLedgerDB 0
+    LgrDB.validate lgrDB genesisLedgerDB BlockCache.empty 0
       (map getHeader (Chain.toOldestFirst chain)) >>= \case
         LgrDB.MaximumRollbackExceeded {} ->
           error "rollback was 0"