Avoid Rollbacks during restart

Currently when there is a restart, db-sync goes back to the latest on disk ledger snapshot. It also deletes all data in the db back to this point as it maintains the property that ledger and db are always at the same point. However in most cases deleting the db is unnecessary and db-sync has to reinsert all the deleted data.

With this pr we change the property: now that the db tip point is never behind the ledger. So on a relatively long rollback received from the chainsync server, db-sync won't delete any db-blocks. It will apply each new block to the ledger state and it will check if the block is already in the db. If it's already there it just goes to the next block. If it's not, it first cleans up the db for any existing blocks with greater or equal BlockNo and then inserts the block. After that the initial property, ledger and db at the same point, is restored.

As an optimization, when db-sync doesn't do a full rollback, ie doesn't delete the db, it will change its ConsistentLevel to DBAheadOfLedger. When the consistency between ledger and db is restored it will change the level back to Consistent. The query I mentioned above only happens with DBAheadOfLedger, so in the normal syncing case no additional queries are executed.

Author: kderme

cardano-db-sync/src/Cardano/DbSync/Api.hs

@@ -8,6 +8,9 @@ module Cardano.DbSync.Api
   ( SyncEnv (..)
   , LedgerEnv (..)
   , SyncOptions (..)
+  , ConsistentLevel (..)
+  , setConsistentLevel
+  , isConsistent
   , mkSyncEnvFromConfig
   , replaceConnection
   , verifyFilePoints
@@ -71,6 +74,7 @@ data SyncEnv = SyncEnv
   , envSystemStart :: !SystemStart
   , envConnString :: ConnectionString
   , envBackend :: !(StrictTVar IO (Strict.Maybe SqlBackend))
+  , envConsistentLevel :: !(StrictTVar IO ConsistentLevel)
   , envOptions :: !SyncOptions
   , envCache :: !Cache
   , envOfflineWorkQueue :: !(TBQueue IO PoolFetchRetry)
@@ -81,6 +85,21 @@ data SyncEnv = SyncEnv
   , envLedger :: !LedgerEnv
   }
 
+data ConsistentLevel = Consistent | DBAheadOfLedger | Unchecked
+  deriving (Show, Eq)
+
+setConsistentLevel :: SyncEnv -> ConsistentLevel -> IO ()
+setConsistentLevel env cst = do
+    logInfo (getTrace env) $ "Setting ConsistencyLevel to " <> textShow cst
+    atomically $ writeTVar (envConsistentLevel env) cst
+
+isConsistent :: SyncEnv -> IO Bool
+isConsistent env = do
+    cst <- readTVarIO (envConsistentLevel env)
+    case cst of
+      Consistent -> pure True
+      _ -> pure False
+
 data SyncOptions = SyncOptions
   { soptExtended :: !Bool
   , soptAbortOnInvalid :: !Bool
@@ -198,6 +217,7 @@ mkSyncEnv trce connSring syncOptions protoInfo nw nwMagic systemStart dir = do
                  (snapshotEveryFollowing syncOptions) (snapshotEveryLagging syncOptions)
   cache <- if soptCache syncOptions then newEmptyCache 100000 else pure uninitiatedCache
   backendVar <- newTVarIO Strict.Nothing
+  consistentLevelVar <- newTVarIO Unchecked
   owq <- newTBQueueIO 100
   orq <- newTBQueueIO 100
   epochVar <- newTVarIO initEpochState
@@ -210,6 +230,7 @@ mkSyncEnv trce connSring syncOptions protoInfo nw nwMagic systemStart dir = do
           , envConnString = connSring
           , envBackend = backendVar
           , envOptions = syncOptions
+          , envConsistentLevel = consistentLevelVar
           , envCache = cache
           , envOfflineWorkQueue = owq
           , envOfflineResultQueue = orq

cardano-db-sync/src/Cardano/DbSync/Cache.hs

@@ -200,24 +200,24 @@ newEmptyCache maCapacity =
 -- NOTE: For 'StakeAddresses' we use a mixed approach. If the rollback is long we just drop
 -- everything, since it is very rare. If not, we query all the StakeAddressesId of blocks
 -- that wil be deleted.
-rollbackCache :: MonadIO m => Cache -> BlockNo -> Word64 -> ReaderT SqlBackend m ()
-rollbackCache cache (BlockNo blkNo) nBlocks =
+rollbackCache :: MonadIO m => Cache -> BlockNo -> Bool -> Word64 -> ReaderT SqlBackend m ()
+rollbackCache cache (BlockNo blkNo) deleteEq nBlocks =
   case cache of
     UninitiatedCache -> pure ()
     Cache ci -> do
       liftIO $ do
         atomically $ writeTVar (cPools ci) Map.empty
         atomically $ modifyTVar (cMultiAssets ci) LRU.cleanup
         atomically $ writeTVar (cPrevBlock ci) Nothing
-      rollbackStakeAddr ci blkNo nBlocks
+      rollbackStakeAddr ci blkNo deleteEq nBlocks
 
-rollbackStakeAddr :: MonadIO m => CacheInternal -> Word64 -> Word64 -> ReaderT SqlBackend m ()
-rollbackStakeAddr ci blkNo nBlocks = do
+rollbackStakeAddr :: MonadIO m => CacheInternal -> Word64 -> Bool -> Word64 -> ReaderT SqlBackend m ()
+rollbackStakeAddr ci blkNo deleteEq nBlocks = do
   if nBlocks > 600
     then liftIO $ atomically $ writeTVar (cStakeCreds ci) Map.empty
     else do
       initMp <- liftIO $ readTVarIO (cStakeCreds ci)
-      stakeAddrIdsToDelete <- DB.queryStakeAddressIdsAfter blkNo
+      stakeAddrIdsToDelete <- DB.queryStakeAddressIdsAfter blkNo deleteEq
       let stakeAddrIdsSetToDelete = Set.fromList stakeAddrIdsToDelete
       let !mp = Map.filter (`Set.notMember` stakeAddrIdsSetToDelete) initMp
       liftIO $ atomically $ writeTVar (cStakeCreds ci) mp

cardano-db-sync/src/Cardano/DbSync/Database.hs

@@ -89,11 +89,17 @@ runActions env actions = do
       case spanDbApply xs of
         ([], DbFinish:_) -> do
             pure Done
-        ([], DbRollBackToPoint pt resultVar : ys) -> do
-            newExceptT $ rollbackToPoint env pt
+        ([], DbRollBackToPoint pt serverTip resultVar : ys) -> do
+            deletedAllBlocks <- newExceptT $ rollbackToPoint env pt serverTip
             points <- if hasLedgerState env
               then lift $ rollbackLedger env pt
               else pure Nothing
+            -- Ledger state always rollbacks at least back to the 'point' given by the Node.
+            -- It needs to rollback even further, if 'points' is not 'Nothing'.
+            -- The db may not rollback to the Node point.
+            case (deletedAllBlocks, points) of
+              (True, Nothing) -> liftIO $ setConsistentLevel env Consistent
+              _ -> liftIO $ setConsistentLevel env DBAheadOfLedger
             blockNo <- lift $ getDbTipBlockNo env
             lift $ atomically $ putTMVar resultVar (points, blockNo)
             dbAction Continue ys

cardano-db-sync/src/Cardano/DbSync/DbAction.hs

@@ -22,12 +22,12 @@ import qualified Control.Concurrent.STM.TBQueue as TBQ
 
 import           Control.Monad.Class.MonadSTM.Strict (StrictTMVar, newEmptyTMVarIO, takeTMVar)
 
-import           Ouroboros.Network.Block (BlockNo)
+import           Ouroboros.Network.Block (BlockNo, Tip (..))
 import qualified Ouroboros.Network.Point as Point
 
 data DbAction
   = DbApplyBlock !CardanoBlock
-  | DbRollBackToPoint !CardanoPoint (StrictTMVar IO (Maybe [CardanoPoint], Point.WithOrigin BlockNo))
+  | DbRollBackToPoint !CardanoPoint !(Tip CardanoBlock) (StrictTMVar IO (Maybe [CardanoPoint], Point.WithOrigin BlockNo))
   | DbFinish
 
 newtype DbActionQueue = DbActionQueue
@@ -39,10 +39,10 @@ mkDbApply = DbApplyBlock
 
 -- | This simulates a synhronous operations, since the thread waits for the db
 -- worker thread to finish the rollback.
-waitRollback :: DbActionQueue -> CardanoPoint -> IO (Maybe [CardanoPoint], Point.WithOrigin BlockNo)
-waitRollback queue point = do
+waitRollback :: DbActionQueue -> CardanoPoint -> Tip CardanoBlock -> IO (Maybe [CardanoPoint], Point.WithOrigin BlockNo)
+waitRollback queue point serverTip = do
     resultVar <- newEmptyTMVarIO
-    atomically $ writeDbActionQueue queue $ DbRollBackToPoint point resultVar
+    atomically $ writeDbActionQueue queue $ DbRollBackToPoint point serverTip resultVar
     atomically $ takeTMVar resultVar
 
 lengthDbActionQueue :: DbActionQueue -> STM Natural

cardano-db-sync/src/Cardano/DbSync/Default.hs

@@ -25,11 +25,10 @@ import           Cardano.DbSync.Era.Shelley.Insert (insertShelleyBlock)
 import           Cardano.DbSync.Era.Shelley.Insert.Epoch (insertPoolDepositRefunds, insertRewards)
 import           Cardano.DbSync.Era.Shelley.Validate (validateEpochRewards)
 import           Cardano.DbSync.Error
-
 import           Cardano.DbSync.LedgerState (ApplyResult (..), LedgerEvent (..),
                    applyBlockAndSnapshot, defaultApplyResult)
 import           Cardano.DbSync.LocalStateQuery
-
+import           Cardano.DbSync.Rollback
 import           Cardano.DbSync.Types
 import           Cardano.DbSync.Util
 
@@ -41,6 +40,7 @@ import           Control.Monad.Logger (LoggingT)
 import           Control.Monad.Trans.Control (MonadBaseControl)
 import           Control.Monad.Trans.Except.Extra (newExceptT)
 
+import qualified Data.ByteString.Short as SBS
 import qualified Data.Map.Strict as Map
 import qualified Data.Set as Set
 import qualified Data.Strict.Maybe as Strict
@@ -49,6 +49,8 @@ import           Database.Persist.SqlBackend.Internal
 import           Database.Persist.SqlBackend.Internal.StatementCache
 
 import           Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
+import qualified Ouroboros.Consensus.HardFork.Combinator as Consensus
+import           Ouroboros.Network.Block (blockHash, blockNo, getHeaderFields)
 
 
 insertListBlocks
@@ -58,38 +60,68 @@ insertListBlocks env blocks = do
     backend <- getBackend env
     DB.runDbIohkLogging backend tracer .
       runExceptT $ do
-        traverse_ (applyAndInsert env) blocks
+        traverse_ (applyAndInsertBlockMaybe env) blocks
   where
     tracer = getTrace env
 
-applyAndInsert
+applyAndInsertBlockMaybe
     :: SyncEnv -> CardanoBlock -> ExceptT SyncNodeError (ReaderT SqlBackend (LoggingT IO)) ()
-applyAndInsert env cblk = do
+applyAndInsertBlockMaybe env cblk = do
     !applyRes <- liftIO mkApplyResult
+    bl <- liftIO $ isConsistent env
+    if bl then
+      -- In the usual case it will be consistent so we don't need to do any queries. Just insert the block
+      insertBlock env cblk applyRes False
+    else do
+      blockIsInDbAlready <- lift (isRight <$> DB.queryBlockHash (SBS.fromShort . Consensus.getOneEraHash $ blockHash cblk))
+      -- If the block is already in db, do nothing. If not, delete all blocks with greater 'BlockNo' or
+      -- equal, insert the block and restore consistency between ledger and db.
+      unless blockIsInDbAlready $ do
+        liftIO . logInfo tracer $
+          mconcat
+           [ "Received block which is not in the db ", textShow (getHeaderFields cblk)
+           , " Time to restore consistency."]
+        rollbackFromBlockNo env (blockNo cblk)
+        insertBlock env cblk applyRes True
+        liftIO $ setConsistentLevel env Consistent
+  where
+    tracer = getTrace env
+
+    mkApplyResult :: IO ApplyResult
+    mkApplyResult = do
+      if hasLedgerState env then
+        applyBlockAndSnapshot (envLedger env) cblk
+      else do
+        slotDetails <- getSlotDetailsNode (envNoLedgerEnv env) (cardanoBlockSlotNo cblk)
+        pure $ defaultApplyResult slotDetails
+
+insertBlock
+    :: SyncEnv -> CardanoBlock -> ApplyResult -> Bool -> ExceptT SyncNodeError (ReaderT SqlBackend (LoggingT IO)) ()
+insertBlock env cblk applyRes logBlock = do
     !epochEvents <- liftIO $ atomically $ generateNewEpochEvents env (apSlotDetails applyRes)
     let !applyResult = applyRes { apEvents = sort $ epochEvents <> apEvents applyRes}
     let !details = apSlotDetails applyResult
     insertLedgerEvents env (sdEpochNo details) (apEvents applyResult)
     insertEpoch details
-    let firstBlockOfEpoch = hasEpochStartEvent (apEvents applyResult)
+    let shouldLog = hasEpochStartEvent (apEvents applyResult) || logBlock
     let isMember poolId = Set.member poolId (apPoolsRegistered applyResult)
     case cblk of
       BlockByron blk ->
-        newExceptT $ insertByronBlock env firstBlockOfEpoch blk details
+        newExceptT $ insertByronBlock env shouldLog blk details
       BlockShelley blk -> newExceptT $
-        insertShelleyBlock env firstBlockOfEpoch (Generic.fromShelleyBlock blk)
+        insertShelleyBlock env shouldLog (Generic.fromShelleyBlock blk)
           details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
       BlockAllegra blk -> newExceptT $
-        insertShelleyBlock env firstBlockOfEpoch (Generic.fromAllegraBlock blk)
+        insertShelleyBlock env shouldLog (Generic.fromAllegraBlock blk)
           details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
       BlockMary blk -> newExceptT $
-        insertShelleyBlock env firstBlockOfEpoch (Generic.fromMaryBlock blk)
+        insertShelleyBlock env shouldLog (Generic.fromMaryBlock blk)
           details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
       BlockAlonzo blk -> newExceptT $
-        insertShelleyBlock env firstBlockOfEpoch (Generic.fromAlonzoBlock (getPrices applyResult) blk)
+        insertShelleyBlock env shouldLog (Generic.fromAlonzoBlock (getPrices applyResult) blk)
           details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
       BlockBabbage blk -> newExceptT $
-        insertShelleyBlock env firstBlockOfEpoch (Generic.fromBabbageBlock (getPrices applyResult) blk)
+        insertShelleyBlock env shouldLog (Generic.fromBabbageBlock (getPrices applyResult) blk)
           details isMember (apNewEpoch applyResult) (apStakeSlice applyResult)
   where
     tracer = getTrace env
@@ -103,14 +135,6 @@ applyAndInsert env cblk = do
       Strict.Nothing | hasLedgerState env -> Just $ Ledger.Prices minBound minBound
       Strict.Nothing -> Nothing
 
-    mkApplyResult :: IO ApplyResult
-    mkApplyResult = do
-      if hasLedgerState env then
-        applyBlockAndSnapshot (envLedger env) cblk
-      else do
-        slotDetails <- getSlotDetailsNode (envNoLedgerEnv env) (cardanoBlockSlotNo cblk)
-        pure $ defaultApplyResult slotDetails
-
 -- -------------------------------------------------------------------------------------------------
 
 insertLedgerEvents

cardano-db-sync/src/Cardano/DbSync/Era/Shelley/Insert.hs

@@ -86,7 +86,7 @@ insertShelleyBlock
     => SyncEnv -> Bool -> Generic.Block -> SlotDetails
     -> IsPoolMember -> Strict.Maybe Generic.NewEpoch -> Generic.StakeSliceRes
     -> ReaderT SqlBackend m (Either SyncNodeError ())
-insertShelleyBlock env firstBlockOfEpoch blk details isMember mNewEpoch stakeSlice = do
+insertShelleyBlock env shouldLog blk details isMember mNewEpoch stakeSlice = do
   runExceptT $ do
     mPhid <- lift $ queryPoolKeyWithCache cache CacheNew $ coerceKeyRole $ Generic.blkSlotLeader blk
     slid <- lift . DB.insertSlotLeader $ Generic.mkSlotLeader (Generic.unKeyHashRaw $ Generic.blkSlotLeader blk) (eitherToMaybe mPhid)
@@ -152,7 +152,7 @@ insertShelleyBlock env firstBlockOfEpoch blk details isMember mNewEpoch stakeSli
   where
     logger :: Bool -> Trace IO a -> a -> IO ()
     logger followingClosely
-      | firstBlockOfEpoch = logInfo
+      | shouldLog = logInfo
       | followingClosely = logInfo
       | unBlockNo (Generic.blkBlockNo blk) `mod` 5000 == 0 = logInfo
       | otherwise = logDebug

cardano-db-sync/src/Cardano/DbSync/Rollback.hs

@@ -4,6 +4,7 @@
 
 module Cardano.DbSync.Rollback
   ( rollbackToPoint
+  , rollbackFromBlockNo
   , unsafeRollback
   ) where
 
@@ -21,58 +22,68 @@ import           Cardano.DbSync.Error
 import           Cardano.DbSync.Types
 import           Cardano.DbSync.Util
 
-import qualified Data.List as List
 import           Database.Persist.Sql (SqlBackend)
 
 import           Ouroboros.Consensus.HardFork.Combinator.AcrossEras (getOneEraHash)
 
 import           Ouroboros.Network.Block
 import           Ouroboros.Network.Point
 
+-- | The decision to delete blocks has been taken and this executes it.
+deleteBlocks :: MonadIO m => SyncEnv -> BlockNo -> Bool -> Word64 -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
+deleteBlocks env blkNo deleteEq nBlocks = do
+    unless (nBlocks == 0) $
+      liftIO . logInfo trce $
+          mconcat
+            [ "Deleting ", textShow nBlocks, " blocks after "
+            , if deleteEq then " or equal to " else ""
+            , textShow blkNo
+            ]
+    -- We need to first cleanup the cache and then delete the blocks from db.
+    lift $ rollbackCache cache blkNo deleteEq nBlocks
+    deleted <- lift $ DB.deleteAfterBlockNo blkNo deleteEq
+    liftIO . logInfo trce $
+                if deleted
+                  then "Blocks deleted"
+                  else "No blocks need to be deleted"
+  where
+    trce :: Trace IO Text
+    trce = getTrace env
+
+    cache :: Cache
+    cache = envCache env
+
+rollbackFromBlockNo :: MonadIO m => SyncEnv -> BlockNo -> ExceptT SyncNodeError (ReaderT SqlBackend m) ()
+rollbackFromBlockNo env blkNo = do
+    nBlocks <- lift $ DB.queryBlockCountAfterBlockNo (unBlockNo blkNo) True
+    deleteBlocks env blkNo True (fromIntegral nBlocks)
+
 -- Rollbacks are done in an Era generic way based on the 'Point' we are
 -- rolling back to.
-rollbackToPoint :: SyncEnv -> CardanoPoint -> IO (Either SyncNodeError ())
-rollbackToPoint env point = do
+rollbackToPoint :: SyncEnv -> CardanoPoint -> Tip CardanoBlock -> IO (Either SyncNodeError Bool)
+rollbackToPoint env point serverTip = do
     backend <- getBackend env
     DB.runDbIohkNoLogging backend $ runExceptT action
   where
     trce :: Trace IO Text
     trce = getTrace env
 
-    cache :: Cache
-    cache = envCache env
-
-    action :: MonadIO m => ExceptT SyncNodeError (ReaderT SqlBackend m) ()
+    action :: MonadIO m => ExceptT SyncNodeError (ReaderT SqlBackend m) Bool
     action = do
-        liftIO . logInfo trce $ "Rolling back to " <> renderPoint point
-        xs <- lift $ slotsToDelete (pointSlot point)
-        unless (null xs) $
-          -- there may be more deleted blocks than slots, because ebbs don't have
-          -- a slot. We can only make an estimation here.
+        blkNo <- liftLookupFail "Rollback.rollbackToPoint.queryBlockNo" $ queryBlockNo point
+        nBlocks <- lift $ DB.queryBlockCountAfterBlockNo (unBlockNo blkNo) False
+        if nBlocks <= 50 || not (hasLedgerState env) then do
+          liftIO . logInfo trce $ "Rolling back to " <> renderPoint point
+          deleteBlocks env blkNo False (fromIntegral nBlocks)
+          pure True
+        else do
           liftIO . logInfo trce $
-              mconcat
-                [ "Deleting ", textShow (length xs), " blocks up to slot "
-                , textShow (unSlotNo $ List.head xs)
-                ]
-        -- We delete the block right after the point we rollback to. This delete
-        -- should cascade to the rest of the chain.
-        blkNo <- liftLookupFail "Rollback.rollbackToPoint" $ queryBlockNo point
-        -- 'length xs' here gives an approximation of the blocks deleted. An approximation
-        -- is good enough, since it is only used to decide on the best policy and is not
-        -- important for correctness.
-        -- We need to first cleanup the cache and then delete the blocks from db.
-        lift $ rollbackCache cache blkNo (fromIntegral $ length xs)
-        deleted <- lift $ DB.deleteAfterBlockNo blkNo
-        liftIO . logInfo trce $
-                    if deleted
-                      then "Blocks deleted"
-                      else "No blocks need to be deleted"
-
-    slotsToDelete :: MonadIO m => WithOrigin SlotNo -> ReaderT SqlBackend m [SlotNo]
-    slotsToDelete wosl =
-      case wosl of
-        Origin -> DB.querySlotNos
-        At sl -> DB.querySlotNosGreaterThan (unSlotNo sl)
+            mconcat
+              [ "Delaying rollback of ", textShow nBlocks, " blocks after "
+              , textShow blkNo, " back to " , renderPoint point
+              , ". Applying blocks up to current node ", textShow serverTip
+              ]
+          pure False
 
     queryBlockNo :: MonadIO m => Point CardanoBlock -> ReaderT SqlBackend m (Either DB.LookupFail BlockNo)
     queryBlockNo pnt =