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

@@ -164,7 +164,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
     traceWith tracer $ TraceOpenEvent StartedOpeningLgrDB
     (ledgerDbGetVolatileSuffix, setGetCurrentChainForLedgerDB) <-
       mkLedgerDbGetVolatileSuffix
-    (lgrDB, replayed) <-
+    (lgrDB, _replayed) <-
       LedgerDB.openDB
         argsLgrDb
         (ImmutableDB.streamAPI immutableDB)
@@ -289,8 +289,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
             , intGarbageCollect = \slot -> getEnv h $ \e -> do
                 Background.garbageCollectBlocks e slot
                 LedgerDB.garbageCollect (cdbLedgerDB e) slot
-            , 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
             }
@@ -301,7 +300,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

@@ -1,13 +1,11 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TupleSections #-}
 
 -- | Background tasks:
 --
@@ -52,7 +50,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
@@ -75,7 +72,7 @@ import qualified Ouroboros.Consensus.Storage.VolatileDB as VolatileDB
 import Ouroboros.Consensus.Util
 import Ouroboros.Consensus.Util.Condense
 import Ouroboros.Consensus.Util.IOLike
-import Ouroboros.Consensus.Util.STM (Watcher (..), forkLinkedWatcher)
+import Ouroboros.Consensus.Util.STM (Watcher (..), blockUntilJust, forkLinkedWatcher)
 import Ouroboros.Network.AnchoredFragment (AnchoredSeq (..))
 import qualified Ouroboros.Network.AnchoredFragment as AF
 
@@ -92,15 +89,13 @@ 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
 
-  ledgerDbTasksTrigger <- newLedgerDbTasksTrigger replayed
+  ledgerDbTasksTrigger <- newLedgerDbTasksTrigger
   !ledgerDbMaintenaceThread <-
     forkLinkedWatcher cdbRegistry "ChainDB.ledgerDbTaskWatcher" $
       ledgerDbTaskWatcher cdb ledgerDbTasksTrigger
@@ -259,20 +254,18 @@ copyToImmutableDBRunner cdb@CDB{..} ledgerDbTasksTrigger gcSchedule fuse = do
   copyAndTrigger :: m ()
   copyAndTrigger = do
     -- Wait for 'cdbChain' to become longer than 'getCurrentChain'.
-    numToWrite <- atomically $ do
+    atomically $ do
       curChain <- icWithoutTime <$> readTVar cdbChain
       curChainVolSuffix <- Query.getCurrentChain cdb
-      let numToWrite = AF.length curChain - AF.length curChainVolSuffix
-      check $ numToWrite > 0
-      return $ fromIntegral numToWrite
+      check $ AF.length curChain > AF.length curChainVolSuffix
 
     -- Copy blocks to ImmutableDB
     --
     -- This is a synchronous operation: when it returns, the blocks have been
     -- copied to disk (though not flushed, necessarily).
     gcSlotNo <- withFuse fuse (copyToImmutableDB cdb)
 
-    triggerLedgerDbTasks ledgerDbTasksTrigger gcSlotNo numToWrite
+    triggerLedgerDbTasks ledgerDbTasksTrigger gcSlotNo
     scheduleGC' gcSlotNo
 
   scheduleGC' :: WithOrigin SlotNo -> m ()
@@ -294,45 +287,20 @@ copyToImmutableDBRunner cdb@CDB{..} ledgerDbTasksTrigger gcSchedule fuse = do
 -- | Trigger for the LedgerDB maintenance tasks, namely whenever the immutable
 -- DB tip slot advances when we finish copying blocks to it.
 newtype LedgerDbTasksTrigger m
-  = LedgerDbTasksTrigger (StrictTVar m LedgerDbTaskState)
+  = LedgerDbTasksTrigger (StrictTVar m (WithOrigin SlotNo))
 
-data LedgerDbTaskState = LedgerDbTaskState
-  { ldbtsImmTip :: !(WithOrigin SlotNo)
-  , ldbtsPrevSnapshotTime :: !(Maybe Time)
-  , ldbtsBlocksSinceLastSnapshot :: !Word64
-  }
-  deriving stock Generic
-  deriving anyclass NoThunks
-
-newLedgerDbTasksTrigger ::
-  IOLike m =>
-  -- | Number of blocks replayed.
-  Word64 ->
-  m (LedgerDbTasksTrigger m)
-newLedgerDbTasksTrigger replayed = LedgerDbTasksTrigger <$> newTVarIO st
- where
-  st =
-    LedgerDbTaskState
-      { ldbtsImmTip = Origin
-      , ldbtsPrevSnapshotTime = Nothing
-      , ldbtsBlocksSinceLastSnapshot = replayed
-      }
+newLedgerDbTasksTrigger :: IOLike m => m (LedgerDbTasksTrigger m)
+newLedgerDbTasksTrigger = LedgerDbTasksTrigger <$> newTVarIO Origin
 
 triggerLedgerDbTasks ::
   forall m.
   IOLike m =>
   LedgerDbTasksTrigger m ->
   -- | New tip of the ImmutableDB.
   WithOrigin SlotNo ->
-  -- | Number of blocks written to the ImmutableDB.
-  Word64 ->
   m ()
-triggerLedgerDbTasks (LedgerDbTasksTrigger varSt) immTip numWritten =
-  atomically $ modifyTVar varSt $ \st ->
-    st
-      { ldbtsImmTip = immTip
-      , ldbtsBlocksSinceLastSnapshot = ldbtsBlocksSinceLastSnapshot st + numWritten
-      }
+triggerLedgerDbTasks (LedgerDbTasksTrigger varSt) =
+  atomically . writeTVar varSt
 
 -- | Run LedgerDB maintenance tasks when 'LedgerDbTasksTrigger' changes.
 --
@@ -344,38 +312,16 @@ ledgerDbTaskWatcher ::
   IOLike m =>
   ChainDbEnv m blk ->
   LedgerDbTasksTrigger m ->
-  Watcher m LedgerDbTaskState (WithOrigin SlotNo)
+  Watcher m SlotNo SlotNo
 ledgerDbTaskWatcher CDB{..} (LedgerDbTasksTrigger varSt) =
   Watcher
-    { wFingerprint = ldbtsImmTip
+    { wFingerprint = id
     , wInitial = Nothing
-    , wReader = readTVar varSt
-    , wNotify =
-        \LedgerDbTaskState
-           { ldbtsImmTip
-           , ldbtsBlocksSinceLastSnapshot = blocksSinceLast
-           , ldbtsPrevSnapshotTime = prevSnapTime
-           } ->
-            whenJust (withOriginToMaybe ldbtsImmTip) $ \slotNo -> do
-              LedgerDB.tryFlush cdbLedgerDB
-
-              now <- getMonotonicTime
-              LedgerDB.SnapCounters
-                { prevSnapshotTime
-                , ntBlocksSinceLastSnap
-                } <-
-                LedgerDB.tryTakeSnapshot
-                  cdbLedgerDB
-                  ((,now) <$> prevSnapTime)
-                  blocksSinceLast
-              atomically $ modifyTVar varSt $ \st ->
-                st
-                  { ldbtsBlocksSinceLastSnapshot =
-                      ldbtsBlocksSinceLastSnapshot st - blocksSinceLast + ntBlocksSinceLastSnap
-                  , ldbtsPrevSnapshotTime = prevSnapshotTime
-                  }
-
-              LedgerDB.garbageCollect cdbLedgerDB slotNo
+    , wReader = blockUntilJust $ withOriginToMaybe <$> readTVar varSt
+    , wNotify = \slotNo -> do
+        LedgerDB.tryFlush cdbLedgerDB
+        LedgerDB.tryTakeSnapshot cdbLedgerDB
+        LedgerDB.garbageCollect cdbLedgerDB slotNo
     }
 
 {-------------------------------------------------------------------------------

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

@@ -147,9 +147,6 @@ module Ouroboros.Consensus.Storage.LedgerDB.API
   , withPrivateTipForker
   , withTipForker
 
-    -- * Snapshots
-  , SnapCounters (..)
-
     -- * Streaming
   , StreamingBackend (..)
   , Yield
@@ -166,7 +163,6 @@ import Codec.CBOR.Decoding
 import Codec.CBOR.Read
 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
@@ -271,18 +267,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.
@@ -429,18 +419,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
 -------------------------------------------------------------------------------}