better index/db intersection computation

Now the index will not always rollback to origin when its tip is not in
the database. It will try to find an intersection point within the
security parameter, and roll back only to that point.

Author: avieth

src/exec/Byron.hs

@@ -22,7 +22,6 @@ import qualified Data.List.NonEmpty as NE
 import Data.Maybe (mapMaybe)
 import qualified Data.Text.Lazy.Builder as Text
 import Data.Typeable (Typeable)
-import Data.Word (Word64)
 import System.Random (StdGen, getStdGen, randomR)
 
 import qualified Cardano.Binary as Binary
@@ -35,11 +34,11 @@ import qualified Pos.Chain.Block as CSL (Block, BlockHeader (..), GenesisBlock,
 import qualified Pos.Infra.Diffusion.Types as CSL
 
 import Ouroboros.Byron.Proxy.Block (Block, ByronBlockOrEBB (..),
-         coerceHashToLegacy, unByronHeaderOrEBB, headerHash)
+         checkpointOffsets, coerceHashToLegacy, unByronHeaderOrEBB, headerHash)
 import Ouroboros.Byron.Proxy.Main
 import Ouroboros.Consensus.Block (Header)
 import Ouroboros.Consensus.Ledger.Byron (ByronGiven)
-import Ouroboros.Consensus.Protocol.Abstract (SecurityParam (maxRollbacks))
+import Ouroboros.Consensus.Protocol.Abstract (SecurityParam)
 import Ouroboros.Network.Block (ChainHash (..), Point, pointHash)
 import qualified Ouroboros.Network.AnchoredFragment as AF
 import qualified Ouroboros.Network.ChainFragment as CF
@@ -134,25 +133,14 @@ download tracer genesisBlock epochSlots securityParam db bp = do
   checkpoints
     :: AF.AnchoredFragment (Header (Block cfg))
     -> [CSL.HeaderHash]
-  checkpoints = mapMaybe pointToHash . AF.selectPoints (fmap fromIntegral offsets)
+  checkpoints = mapMaybe pointToHash .
+    AF.selectPoints (fmap fromIntegral (checkpointOffsets securityParam))
 
   pointToHash :: Point (Header (Block cfg)) -> Maybe CSL.HeaderHash
   pointToHash pnt = case pointHash pnt of
     GenesisHash    -> Nothing
     BlockHash hash -> Just $ coerceHashToLegacy hash
 
-  -- Offsets for selectPoints. Defined in the same way as for the Shelley
-  -- chain sync client: fibonacci numbers including 0 and k.
-  offsets :: [Word64]
-  offsets = 0 : foldr includeK ([] {- this is never forced -}) (tail fibs)
-
-  includeK :: Word64 -> [Word64] -> [Word64]
-  includeK w ws | w >= k    = [k]
-                | otherwise = w : ws
-
-  fibs :: [Word64]
-  fibs = 1 : 1 : zipWith (+) fibs (tail fibs)
-
   streamer :: CSL.HeaderHash -> CSL.StreamBlocks CSL.Block IO CSL.HeaderHash
   streamer tipHash = CSL.StreamBlocks
     { CSL.streamBlocksMore = \blocks -> do
@@ -181,9 +169,6 @@ download tracer genesisBlock epochSlots securityParam db bp = do
     let (idx, rndGen') = randomR (0, NE.length ne - 1) rndGen
     in  (ne NE.!! idx, rndGen')
 
-  k :: Word64
-  k = maxRollbacks securityParam
-
 recodeBlockOrFail
   :: Cardano.EpochSlots
   -> (forall x . Binary.DecoderError -> IO x)

src/lib/Ouroboros/Byron/Proxy/Block.hs

@@ -20,9 +20,12 @@ module Ouroboros.Byron.Proxy.Block
   , coerceHashToLegacy
   , headerHash
   , isEBB
+  , checkpointOffsets
   ) where
 
 import qualified Codec.CBOR.Write as CBOR (toStrictByteString)
+import qualified Data.ByteString.Lazy as Lazy
+import Data.Word (Word64)
 
 import qualified Pos.Chain.Block as CSL (HeaderHash)
 import qualified Pos.Crypto.Hashing as Legacy (AbstractHash (..))
@@ -34,6 +37,7 @@ import Cardano.Crypto.Hashing (AbstractHash (..))
 import qualified Ouroboros.Consensus.Block as Consensus (GetHeader (..))
 import Ouroboros.Consensus.Ledger.Byron (ByronBlockOrEBB (..),
          pattern ByronHeaderOrEBB, encodeByronBlock, unByronHeaderOrEBB)
+import Ouroboros.Consensus.Protocol.Abstract (SecurityParam (..))
 
 -- For type instance HeaderHash (Header blk) = HeaderHash blk
 -- Anyone who imports this module will almost certainly want that instance.
@@ -70,3 +74,16 @@ isEBB :: Block cfg -> Maybe Cardano.HeaderHash
 isEBB blk = case unByronBlockOrEBB blk of
   Cardano.ABOBBlock    _ -> Nothing
   Cardano.ABOBBoundary _ -> Just $ headerHash (Consensus.getHeader blk)
+
+-- | Compute the offsets for use by ChainFragment.selectPoints or
+-- AnchoredFragment.selectPoints for some security parameter k. It uses
+-- fibonacci numbers with 0 and k as endpoints, and the duplicate 1 at the
+-- start removed.
+checkpointOffsets :: SecurityParam -> [Word64]
+checkpointOffsets (SecurityParam k) = 0 : foldr includeK ([] {- this is never forced -}) (tail fibs)
+  where
+    includeK :: Word64 -> [Word64] -> [Word64]
+    includeK w ws | w >= k    = [k]
+                  | otherwise = w : ws
+    fibs :: [Word64]
+    fibs = 1 : 1 : zipWith (+) fibs (tail fibs)

src/lib/Ouroboros/Byron/Proxy/Index/ChainDB.hs

@@ -6,13 +6,16 @@ module Ouroboros.Byron.Proxy.Index.ChainDB
   ) where
 
 import Control.Exception (bracket)
+import Data.Word (Word64)
 
+import Ouroboros.Byron.Proxy.Block (checkpointOffsets)
 import Ouroboros.Byron.Proxy.Index.Types (Index)
 import qualified Ouroboros.Byron.Proxy.Index.Types as Index
 import Ouroboros.Consensus.Block (GetHeader (Header))
 import Ouroboros.Consensus.Util.ResourceRegistry (ResourceRegistry)
+import Ouroboros.Consensus.Protocol.Abstract (SecurityParam)
 import Ouroboros.Network.Block (ChainUpdate (..), Point (..))
-import Ouroboros.Network.Point (WithOrigin (Origin))
+import Ouroboros.Network.Point (WithOrigin (Origin), block)
 import Ouroboros.Storage.ChainDB.API (ChainDB, Reader)
 import qualified Ouroboros.Storage.ChainDB.API as ChainDB
 
@@ -21,7 +24,7 @@ trackReaderBlocking
   :: ( Monad m )
   => Index m (Header blk)
   -> Reader m blk (Header blk)
-  -> m x
+  -> m void
 trackReaderBlocking idx reader = do
   instruction <- ChainDB.readerInstructionBlocking reader
   case instruction of
@@ -50,42 +53,69 @@ trackReader idx reader = do
       trackReader idx reader
     Nothing -> pure ()
 
--- | Have an Index track a ChainDB using its Reader API. You probably want to
--- race this with some other thread that runs your application.
---
--- If the ChainDB does not contain the tip of the Index, then the whole index
--- will be rebuilt.
+-- | Have an Index track a ChainDB using its Reader API for the duration of
+-- some monadic action.
 --
 -- It will spawn a thread to do the index updates. This must be the only
 -- index writer. It is run by `race` with the action, so exceptions in either
 -- the action or the writer thread will be re-thrown here.
 --
--- If the tip of the index is not in the ChainDB, then the entire index will be
--- rebuilt. This is not ideal: there may be an intersection. TODO would be
--- better to check the newest slot older than `k` back from tip of index, and
--- go from there.
+-- If the tip of the index is in the ChainDB, then no work must be done in the
+-- beginning. But if it's not in the ChainDB, there will have to be a rollback
+-- on the index. The SecurityParam k is used to decide how far back to try. If
+-- Only index entries at most k slots old will be checked against the
+-- ChainDB. If none are in it, then the entire index will be rebuild (rollback
+-- to Origin).
 trackChainDB
-  :: forall blk void .
+  :: forall blk t .
      ResourceRegistry IO
   -> Index IO (Header blk)
   -> ChainDB IO blk
-  -> IO void
-trackChainDB rr idx cdb = bracket acquireReader releaseReader $ \rdr -> do
-  tipPoint <- Index.tip idx
-  mPoint <- ChainDB.readerForward rdr [Point tipPoint]
-  -- `readerForward` docs say that if we get `Nothing`, the next reader
-  -- instruction may not be a rollback, so we'll manually roll the index
-  -- back. It's assumed the read pointer will be at origin (nothing else
-  -- would make sense).
-  case mPoint of
-    Nothing -> Index.rollbackward idx Origin
-    Just _  -> pure ()
-  -- First, block until the index is caught up to the tip ...
-  trackReader idx rdr
-  -- ... then attempt to stay in sync.
-  trackReaderBlocking idx rdr
+  -> SecurityParam
+  -> IO t
+  -> IO t
+trackChainDB rr idx cdb k act = bracket acquireReader releaseReader $ \rdr -> do
+    checkpoints <- Index.streamFromTip idx checkpointsFold
+    mPoint <- ChainDB.readerForward rdr checkpoints
+    case mPoint of
+      -- `readerForward` docs say that the next instruction will be a rollback,
+      -- so we don't have to do anything here; the call to `trackReader` will
+      -- do what needs to be done.
+      Just _  -> pure ()
+      -- `readerForward` docs say that if we get `Nothing`, the next reader
+      -- instruction may not be a rollback, so we'll manually roll the index
+      -- back. It's assumed the read pointer will be at origin (nothing else
+      -- would make sense).
+      Nothing -> Index.rollbackward idx Origin
+    -- First, block until the index is caught up to the tip ...
+    trackReader idx rdr
+    -- ... then attempt to stay in sync.
+    outcome <- race (trackReaderBlocking idx rdr) act
+    case outcome of
+      Left impossible -> impossible
+      Right t -> pure t
   where
-  acquireReader :: IO (Reader IO blk (Header blk))
-  acquireReader = ChainDB.newHeaderReader cdb rr
-  releaseReader :: Reader IO blk (Header blk) -> IO ()
-  releaseReader = ChainDB.readerClose
+    acquireReader :: IO (Reader IO blk (Header blk))
+    acquireReader = ChainDB.newHeaderReader cdb rr
+    releaseReader :: Reader IO blk (Header blk) -> IO ()
+    releaseReader = ChainDB.readerClose
+
+    checkpointsFold :: Index.Fold (Header blk) [Point blk]
+    checkpointsFold = checkpointsFoldN 0 (checkpointOffsets k)
+
+    -- Count up from 0 on the first parameter. Whenever it coincides with the
+    -- head of the second parameter (an increasing list) include that point.
+    -- Stop when the second list is empty.
+    -- Since checkpointsFold always includes the paramater k, the k'th entry
+    -- in the index will always be in here, unless the index is shorter
+    -- than k. This block is _at least_ k slots behind the DB, so if it's not
+    -- in the DB then the index is way out of date.
+    checkpointsFoldN
+      :: Word64
+      -> [Word64]
+      -> Index.Fold (Header blk) [Point blk]
+    checkpointsFoldN _ []       = Index.Stop []
+    checkpointsFoldN w (o : os) = Index.More [] $ \slotNo hash ->
+      if w == o
+      then fmap ((:) (Point (block slotNo hash))) (checkpointsFoldN (w+1) os)
+      else checkpointsFoldN (w+1) (o : os)

src/lib/Ouroboros/Byron/Proxy/Index/Sqlite.hs

@@ -46,10 +46,11 @@ data TraceEvent where
 -- (BEGIN TRANSACTION).
 index :: EpochSlots -> Tracer IO TraceEvent -> Connection -> Index IO (Header (Block cfg))
 index epochSlots tracer conn = Index
-  { Index.lookup = sqliteLookup epochSlots conn
-  , tip          = sqliteTip epochSlots conn
-  , rollforward  = sqliteRollforward epochSlots tracer conn
-  , rollbackward = sqliteRollbackward epochSlots tracer conn
+  { Index.lookup  = sqliteLookup epochSlots conn
+  , tip           = sqliteTip epochSlots conn
+  , streamFromTip = sqliteStreamFromTip epochSlots conn
+  , rollforward   = sqliteRollforward epochSlots tracer conn
+  , rollbackward  = sqliteRollbackward epochSlots tracer conn
   }
 
 -- | Open a new or existing SQLite database. If new, it will set up the schema.
@@ -145,6 +146,24 @@ createTable conn = Sql.execute_ conn sql_create_table
 createIndex :: Sql.Connection -> IO ()
 createIndex conn = Sql.execute_ conn sql_create_index
 
+convertHashBlob :: ByteString -> IO HeaderHash
+convertHashBlob blob = case digestFromByteString blob of
+  Just hh -> pure (AbstractHash hh)
+  Nothing -> throwIO $ InvalidHash blob
+
+-- | Convert the database encoding of relative slot to the offset in an
+-- epoch. The header hash is taken for error-reporting purposes.
+offsetInEpoch :: HeaderHash -> Int -> IO Word64
+offsetInEpoch hh i
+  | i == -1 = pure 0
+  | i >= 0  = pure $ fromIntegral i
+  | otherwise = throwIO $ InvalidRelativeSlot hh i
+
+toAbsoluteSlot :: EpochSlots -> HeaderHash -> Word64 -> Int -> IO SlotNo
+toAbsoluteSlot epochSlots hh epochNo slotInt = do
+  offset <- offsetInEpoch hh slotInt
+  pure $ SlotNo $ unEpochSlots epochSlots * epochNo + offset
+
 -- | The tip is the entry with the highest epoch and slot pair.
 sql_get_tip :: Query
 sql_get_tip =
@@ -157,18 +176,36 @@ sqliteTip epochSlots conn = do
    case rows of
      [] -> pure Origin
      ((hhBlob, epoch, slotInt) : _) -> do
-       hh <- case digestFromByteString hhBlob of
-         Just hh -> pure (AbstractHash hh)
-         Nothing -> throwIO $ InvalidHash hhBlob
-       offsetInEpoch :: Word64 <-
-         if slotInt == -1
-         then pure 0
-         else if slotInt >= 0
-         then pure $ fromIntegral slotInt
-         else throwIO $ InvalidRelativeSlot hh slotInt
-       let slotNo = SlotNo $ unEpochSlots epochSlots * epoch + offsetInEpoch
+       hh <- convertHashBlob hhBlob
+       slotNo <- toAbsoluteSlot epochSlots hh epoch slotInt
        pure $ At $ Point.Block slotNo hh
 
+sql_get_all :: Query
+sql_get_all =
+  "SELECT header_hash, epoch, slot FROM block_index\
+  \ ORDER BY epoch DESC, slot DESC;"
+
+-- | Stream rows from the tip by using the prepared statement and nextRow
+-- API.
+sqliteStreamFromTip
+  :: EpochSlots
+  -> Sql.Connection
+  -> Index.Fold (Header (Block cfg)) t
+  -> IO t
+sqliteStreamFromTip epochSlots conn fold = Sql.withStatement conn sql_get_all $ \stmt ->
+    go stmt fold
+  where
+    go stmt step = case step of
+      Index.Stop t -> pure t
+      Index.More t k -> do
+        next <- Sql.nextRow stmt
+        case next of
+          Nothing -> pure t
+          Just (hhBlob, epoch, slotInt) -> do
+            hh <- convertHashBlob hhBlob
+            slotNo <- toAbsoluteSlot epochSlots hh epoch slotInt
+            go stmt (k slotNo hh)
+
 sql_get_hash :: Query
 sql_get_hash =
   "SELECT epoch, slot FROM block_index\
@@ -182,13 +219,8 @@ sqliteLookup epochSlots conn hh@(AbstractHash digest) = do
   case rows of
     [] -> pure Nothing
     ((epoch, slotInt) : _) -> do
-      offsetInEpoch :: Word64 <-
-        if slotInt == -1
-        then pure 0
-        else if slotInt >= 0
-        then pure $ fromIntegral slotInt
-        else throwIO $ InvalidRelativeSlot hh slotInt
-      pure $ Just $ SlotNo $ unEpochSlots epochSlots * epoch + offsetInEpoch
+      slotNo <- toAbsoluteSlot epochSlots hh epoch slotInt
+      pure $ Just slotNo
 
 -- | Note that there is a UNIQUE constraint. This will fail if a duplicate
 -- entry is inserted.

src/lib/Ouroboros/Byron/Proxy/Index/Types.hs

@@ -1,7 +1,9 @@
 {-# LANGUAGE GADTSyntax #-}
+{-# LANGUAGE RankNTypes #-}
 
 module Ouroboros.Byron.Proxy.Index.Types
   ( Index (..)
+  , Fold (..)
   ) where
 
 import Ouroboros.Network.Block (HeaderHash, SlotNo (..))
@@ -19,19 +21,32 @@ import qualified Ouroboros.Network.Point as Point (Block)
 data Index m header = Index
   { -- | Lookup the epoch number and relative slot for a given header hash.
     -- `Nothing` means it's not in the index.
-    lookup       :: HeaderHash header -> m (Maybe SlotNo)
+    lookup        :: HeaderHash header -> m (Maybe SlotNo)
     -- | Check the current tip. `Nothing` means the index is empty. Otherwise,
     -- you get the point and also its header hash.
-  , tip          :: m (WithOrigin (Point.Block SlotNo (HeaderHash header)))
+  , tip           :: m (WithOrigin (Point.Block SlotNo (HeaderHash header)))
+    -- | Lazily fold over all entries in the index beginning at the tip.
+  , streamFromTip :: forall t . Fold header t -> m t
     -- | Extend the index with a new entry. The point must be newer than
     -- the latest point in the index (current tip). Whether this is checked
     -- or enforced depends upon the implementation.
-  , rollforward  :: header -> m ()
+  , rollforward   :: header -> m ()
     -- | Roll back to a given point, making it the tip of the index.
     -- TPoint is used because you can rollback to the origin, clearing the
     -- index.
     -- An index implementation need not actually use the header hash here.
     -- It could or could not check that the point actually corresponds to the
     -- entry at that hash.
-  , rollbackward :: WithOrigin (Point.Block SlotNo (HeaderHash header)) -> m ()
+  , rollbackward  :: WithOrigin (Point.Block SlotNo (HeaderHash header)) -> m ()
   }
+
+-- | Defined for use in streamFromTip. A fold over the slot number and header
+-- hash pairs in an index.
+data Fold header t where
+  Stop :: t -> Fold header t
+  -- | First argument is for when there are no more entries in the index.
+  More :: t -> (SlotNo -> HeaderHash header -> Fold header t) -> Fold header t
+
+instance Functor (Fold header) where
+  fmap f (Stop t) = Stop (f t)
+  fmap f (More eof k) = More (f eof) (\slot hash -> fmap f (k slot hash))