LedgerDB: implement predictable snapshotting

Author: amesgen

ouroboros-consensus/changelog.d/20250630_100300_alexander.esgen_predictable_snapshots.md

@@ -0,0 +1,6 @@
+### Breaking
+
+- LedgerDB: implemented *predictable* snapshots, i.e. different nodes with the
+  same configuration will now create snapshots for the same slots.
+
+  See 'SnapshotPolicyArgs' for more details.

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

@@ -160,7 +160,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
   (chainDB, testing, env) <- lift $ do
     traceWith tracer $ TraceOpenEvent (OpenedVolatileDB maxSlot)
     traceWith tracer $ TraceOpenEvent StartedOpeningLgrDB
-    (lgrDB, replayed) <-
+    (lgrDB, _replayed) <-
       LedgerDB.openDB
         argsLgrDb
         (ImmutableDB.streamAPI immutableDB)
@@ -279,8 +279,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
           Internal
             { intCopyToImmutableDB = getEnv h (withFuse copyTestFuse . Background.copyToImmutableDB)
             , intGarbageCollect = getEnv1 h Background.garbageCollect
-            , intTryTakeSnapshot = getEnv h $ \env' ->
-                void $ LedgerDB.tryTakeSnapshot (cdbLedgerDB env') Nothing maxBound
+            , intTryTakeSnapshot = getEnv h $ LedgerDB.tryTakeSnapshot . cdbLedgerDB
             , intAddBlockRunner = getEnv h (Background.addBlockRunner addBlockTestFuse)
             , intKillBgThreads = varKillBgThreads
             }
@@ -291,7 +290,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
           (castPoint $ AF.anchorPoint chain)
           (castPoint $ AF.headPoint chain)
 
-    when launchBgTasks $ Background.launchBgTasks env replayed
+    when launchBgTasks $ Background.launchBgTasks env
 
     return (chainDB, testing, env)
 

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

@@ -6,7 +6,6 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TupleSections #-}
 
 -- | Background tasks:
 --
@@ -53,7 +52,6 @@ import Data.Sequence.Strict (StrictSeq (..))
 import qualified Data.Sequence.Strict as Seq
 import Data.Time.Clock
 import Data.Void (Void)
-import Data.Word
 import GHC.Generics (Generic)
 import GHC.Stack (HasCallStack)
 import Ouroboros.Consensus.Block
@@ -93,10 +91,8 @@ launchBgTasks ::
   , HasHardForkHistory blk
   ) =>
   ChainDbEnv m blk ->
-  -- | Number of immutable blocks replayed on ledger DB startup
-  Word64 ->
   m ()
-launchBgTasks cdb@CDB{..} replayed = do
+launchBgTasks cdb@CDB{..} = do
   !addBlockThread <-
     launch "ChainDB.addBlockRunner" $
       addBlockRunner cdbChainSelFuse cdb
@@ -107,7 +103,7 @@ launchBgTasks cdb@CDB{..} replayed = do
         garbageCollect cdb
   !copyAndSnapshotThread <-
     launch "ChainDB.copyAndSnapshotRunner" $
-      copyAndSnapshotRunner cdb gcSchedule replayed cdbCopyFuse
+      copyAndSnapshotRunner cdb gcSchedule cdbCopyFuse
   atomically $
     writeTVar cdbKillBgThreads $
       sequence_ [addBlockThread, gcThread, copyAndSnapshotThread]
@@ -241,30 +237,22 @@ copyAndSnapshotRunner ::
   ) =>
   ChainDbEnv m blk ->
   GcSchedule m ->
-  -- | Number of immutable blocks replayed on ledger DB startup
-  Word64 ->
   Fuse m ->
   m Void
-copyAndSnapshotRunner cdb@CDB{..} gcSchedule replayed fuse = do
+copyAndSnapshotRunner cdb@CDB{..} gcSchedule fuse = do
   -- this first flush will persist the differences that come from the initial
   -- chain selection.
   LedgerDB.tryFlush cdbLedgerDB
-  loop =<< LedgerDB.tryTakeSnapshot cdbLedgerDB Nothing replayed
+  forever copyAndSnapshot
  where
   SecurityParam k = configSecurityParam cdbTopLevelConfig
 
-  loop :: LedgerDB.SnapCounters -> m Void
-  loop counters = do
-    let LedgerDB.SnapCounters
-          { prevSnapshotTime
-          , ntBlocksSinceLastSnap
-          } = counters
-
+  copyAndSnapshot :: m ()
+  copyAndSnapshot = do
     -- Wait for the chain to grow larger than @k@
-    numToWrite <- atomically $ do
+    atomically $ do
       curChain <- icWithoutTime <$> readTVar cdbChain
       check $ fromIntegral (AF.length curChain) > unNonZero k
-      return $ fromIntegral (AF.length curChain) - unNonZero k
 
     -- Copy blocks to ImmutableDB
     --
@@ -273,16 +261,13 @@ copyAndSnapshotRunner cdb@CDB{..} gcSchedule replayed fuse = do
     gcSlotNo <- withFuse fuse (copyToImmutableDB cdb)
     scheduleGC' gcSlotNo
 
+    LedgerDB.tryTakeSnapshot cdbLedgerDB
+
     -- See the Haddocks above as for why we garbage-collect the LedgerDB already
     -- here (instead of as part of the scheduled GC).
     whenJust (withOriginToMaybe gcSlotNo) $ LedgerDB.garbageCollect cdbLedgerDB
     LedgerDB.tryFlush cdbLedgerDB
 
-    now <- getMonotonicTime
-    let ntBlocksSinceLastSnap' = ntBlocksSinceLastSnap + numToWrite
-
-    loop =<< LedgerDB.tryTakeSnapshot cdbLedgerDB ((,now) <$> prevSnapshotTime) ntBlocksSinceLastSnap'
-
   scheduleGC' :: WithOrigin SlotNo -> m ()
   scheduleGC' Origin = return ()
   scheduleGC' (NotOrigin slotNo) =

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

@@ -149,9 +149,6 @@ module Ouroboros.Consensus.Storage.LedgerDB.API
   , withPrivateTipForker
   , withTipForker
 
-    -- * Snapshots
-  , SnapCounters (..)
-
     -- * Testing
   , TestInternals (..)
   , TestInternals'
@@ -160,7 +157,6 @@ module Ouroboros.Consensus.Storage.LedgerDB.API
 
 import Codec.Serialise
 import qualified Control.Monad as Monad
-import Control.Monad.Class.MonadTime.SI
 import Control.Monad.Except
 import Control.ResourceRegistry
 import Control.Tracer
@@ -262,18 +258,12 @@ data LedgerDB m l blk = LedgerDB
   --  * The set of previously applied points.
   , tryTakeSnapshot ::
       l ~ ExtLedgerState blk =>
-      Maybe (Time, Time) ->
-      Word64 ->
-      m SnapCounters
+      m ()
   -- ^ If the provided arguments indicate so (based on the SnapshotPolicy with
   -- which this LedgerDB was opened), take a snapshot and delete stale ones.
   --
-  -- The arguments are:
-  --
-  -- - If a snapshot has been taken already, the time at which it was taken
-  --   and the current time.
-  --
-  -- - How many blocks have been processed since the last snapshot.
+  -- For V1, this must not be called concurrently with 'garbageCollect' and/or
+  -- 'tryFlush'.
   , tryFlush :: m ()
   -- ^ Flush V1 in-memory LedgerDB state to disk, if possible. This is a no-op
   -- for implementations that do not need an explicit flush function.
@@ -420,18 +410,6 @@ getReadOnlyForker ::
   m (Either GetForkerError (ReadOnlyForker m l blk))
 getReadOnlyForker ldb rr pt = fmap readOnlyForker <$> getForkerAtTarget ldb rr pt
 
-{-------------------------------------------------------------------------------
-  Snapshots
--------------------------------------------------------------------------------}
-
--- | Counters to keep track of when we made the last snapshot.
-data SnapCounters = SnapCounters
-  { prevSnapshotTime :: !(Maybe Time)
-  -- ^ When was the last time we made a snapshot
-  , ntBlocksSinceLastSnap :: !Word64
-  -- ^ How many blocks have we processed since the last snapshot
-  }
-
 {-------------------------------------------------------------------------------
   Initialization
 -------------------------------------------------------------------------------}