LedgerDB: implement predictable snapshotting

ouroboros-consensus-cardano/src/unstable-cardano-tools/Cardano/Tools/DBAnalyser/Run.hs

@@ -64,7 +64,7 @@ openLedgerDB ::
     , LedgerDB.TestInternals' IO blk
     )
 openLedgerDB [email protected]{LedgerDB.lgrFlavorArgs = LedgerDB.LedgerDbFlavorArgsV1 bss} = do
-  (ledgerDB, _, intLedgerDB) <-
+  (ledgerDB, intLedgerDB) <-
     LedgerDB.openDBInternal
       lgrDbArgs
       ( LedgerDB.V1.mkInitDb
@@ -76,7 +76,7 @@ openLedgerDB [email protected]{LedgerDB.lgrFlavorArgs = LedgerDB.L
       genesisPoint
   pure (ledgerDB, intLedgerDB)
 openLedgerDB [email protected]{LedgerDB.lgrFlavorArgs = LedgerDB.LedgerDbFlavorArgsV2 args} = do
-  (ledgerDB, _, intLedgerDB) <-
+  (ledgerDB, intLedgerDB) <-
     LedgerDB.openDBInternal
       lgrDbArgs
       ( LedgerDB.V2.mkInitDb

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/ouroboros-consensus.cabal

@@ -685,6 +685,7 @@ test-suite storage-test
     Test.Ouroboros.Storage.ChainDB.FollowerPromptness
     Test.Ouroboros.Storage.ChainDB.GcSchedule
     Test.Ouroboros.Storage.ChainDB.Iterator
+    Test.Ouroboros.Storage.ChainDB.LedgerSnapshots
     Test.Ouroboros.Storage.ChainDB.Model
     Test.Ouroboros.Storage.ChainDB.Model.Test
     Test.Ouroboros.Storage.ChainDB.Paths
@@ -698,7 +699,6 @@ test-suite storage-test
     Test.Ouroboros.Storage.ImmutableDB.StateMachine
     Test.Ouroboros.Storage.LedgerDB
     Test.Ouroboros.Storage.LedgerDB.Serialisation
-    Test.Ouroboros.Storage.LedgerDB.SnapshotPolicy
     Test.Ouroboros.Storage.LedgerDB.Snapshots
     Test.Ouroboros.Storage.LedgerDB.StateMachine
     Test.Ouroboros.Storage.LedgerDB.StateMachine.TestBlock
@@ -721,7 +721,7 @@ test-suite storage-test
     bytestring,
     cardano-binary,
     cardano-ledger-binary:testlib,
-    cardano-ledger-core:{cardano-ledger-core, testlib},
+    cardano-ledger-core:cardano-ledger-core,
     cardano-slotting:{cardano-slotting, testlib},
     cardano-strict-containers,
     cborg,

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 <-
       LedgerDB.openDB
         argsLgrDb
         (ImmutableDB.streamAPI immutableDB)
@@ -281,8 +281,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
             }
@@ -293,7 +292,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:
 --
@@ -53,7 +51,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
@@ -76,7 +73,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
 
@@ -93,15 +90,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,18 +254,17 @@ copyToImmutableDBRunner cdb@CDB{..} ledgerDbTasksTrigger gcSchedule fuse = do
   copyAndTrigger :: m ()
   copyAndTrigger = 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
     --
     -- 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 ()
@@ -292,45 +286,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.
 --
@@ -342,38 +311,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.hs

@@ -18,7 +18,6 @@ module Ouroboros.Consensus.Storage.LedgerDB
   ) where
 
 import Data.Functor.Contravariant ((>$<))
-import Data.Word
 import Ouroboros.Consensus.Block
 import Ouroboros.Consensus.HardFork.Abstract
 import Ouroboros.Consensus.Ledger.Inspect
@@ -58,7 +57,7 @@ openDB ::
   Point blk ->
   -- | How to get blocks from the ChainDB
   ResolveBlock m blk ->
-  m (LedgerDB' m blk, Word64)
+  m (LedgerDB' m blk)
 openDB
   args
   stream
@@ -94,11 +93,9 @@ doOpenDB ::
   InitDB db m blk ->
   StreamAPI m blk blk ->
   Point blk ->
-  m (LedgerDB' m blk, Word64)
+  m (LedgerDB' m blk)
 doOpenDB args initDb stream replayGoal =
-  f <$> openDBInternal args initDb stream replayGoal
- where
-  f (ldb, replayCounter, _) = (ldb, replayCounter)
+  fst <$> openDBInternal args initDb stream replayGoal
 
 -- | Open the ledger DB and expose internals for testing purposes
 openDBInternal ::
@@ -111,10 +108,10 @@ openDBInternal ::
   InitDB db m blk ->
   StreamAPI m blk blk ->
   Point blk ->
-  m (LedgerDB' m blk, Word64, TestInternals' m blk)
+  m (LedgerDB' m blk, TestInternals' m blk)
 openDBInternal args@(LedgerDbArgs{lgrHasFS = SomeHasFS fs}) initDb stream replayGoal = do
   createDirectoryIfMissing fs True (mkFsPath [])
-  (_initLog, db, replayCounter) <-
+  (_initLog, db) <-
     initialize
       replayTracer
       snapTracer
@@ -125,7 +122,7 @@ openDBInternal args@(LedgerDbArgs{lgrHasFS = SomeHasFS fs}) initDb stream replay
       initDb
       lgrStartSnapshot
   (ledgerDb, internal) <- mkLedgerDb initDb db
-  return (ledgerDb, replayCounter, internal)
+  return (ledgerDb, internal)
  where
   LedgerDbArgs
     { lgrConfig