Modify abstract interface for LedgerDB/ChainDB for sublibraries

jasagredo · jasagredo · commit e48572a0dd4f · 2025-09-18T14:13:44.000+02:00
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Args.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Args.hs
@@ -29,6 +29,7 @@ import Ouroboros.Consensus.Block
 import Ouroboros.Consensus.Config
 import Ouroboros.Consensus.Ledger.Abstract
 import Ouroboros.Consensus.Ledger.Extended
+import Ouroboros.Consensus.Ledger.SupportsProtocol
 import Ouroboros.Consensus.Protocol.Abstract
 import Ouroboros.Consensus.Storage.ChainDB.API
   ( GetLoEFragment
@@ -38,9 +39,11 @@ import Ouroboros.Consensus.Storage.ChainDB.Impl.Types
   ( TraceEvent (..)
   )
 import qualified Ouroboros.Consensus.Storage.ImmutableDB as ImmutableDB
-import Ouroboros.Consensus.Storage.LedgerDB (LedgerDbFlavorArgs)
+import Ouroboros.Consensus.Storage.LedgerDB (LedgerDbBackendArgs)
 import qualified Ouroboros.Consensus.Storage.LedgerDB as LedgerDB
 import Ouroboros.Consensus.Storage.LedgerDB.Snapshots
+import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.Backend as LedgerDB
+import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.InMemory as InMemory
 import qualified Ouroboros.Consensus.Storage.VolatileDB as VolatileDB
 import Ouroboros.Consensus.Util.Args
 import Ouroboros.Consensus.Util.IOLike
@@ -131,13 +134,17 @@ defaultSpecificArgs =
 -- and must therefore be set explicitly.
 defaultArgs ::
   forall m blk.
-  Monad m =>
+  ( IOLike m
+  , LedgerDB.LedgerDbSerialiseConstraints blk
+  , LedgerSupportsProtocol blk
+  , LedgerDB.LedgerSupportsInMemoryLedgerDB (LedgerState blk)
+  ) =>
   Incomplete ChainDbArgs m blk
 defaultArgs =
   ChainDbArgs
     ImmutableDB.defaultArgs
     VolatileDB.defaultArgs
-    LedgerDB.defaultArgs
+    (LedgerDB.defaultArgs $ LedgerDB.SomeBackendArgs InMemory.InMemArgs)
     defaultSpecificArgs
 
 ensureValidateAll ::
@@ -169,7 +176,7 @@ completeChainDbArgs ::
   (RelativeMountPoint -> SomeHasFS m) ->
   -- | Volatile  FS, see 'NodeDatabasePaths'
   (RelativeMountPoint -> SomeHasFS m) ->
-  Complete LedgerDbFlavorArgs m ->
+  LedgerDbBackendArgs m blk ->
   -- | A set of incomplete arguments, possibly modified wrt @defaultArgs@
   Incomplete ChainDbArgs m blk ->
   Complete ChainDbArgs m blk
@@ -206,7 +213,7 @@ completeChainDbArgs
                 LedgerDB.configLedgerDb
                   cdbsTopLevelConfig
                   (LedgerDB.ledgerDbCfgComputeLedgerEvents $ LedgerDB.lgrConfig (cdbLgrDbArgs defArgs))
-            , LedgerDB.lgrFlavorArgs = flavorArgs
+            , LedgerDB.lgrBackendArgs = flavorArgs
             , LedgerDB.lgrRegistry = registry
             }
       , cdbsArgs =
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/ChainSel.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/ChainSel.hs
@@ -1094,7 +1094,7 @@ chainSelection chainSelEnv rr chainDiffs =
         -- Just in case, explicitly yield to ensure that a capability (by
         -- default, the node uses just two) has the opportunity to switch
         -- to a ChainSync server thread.
-        yield
+        Ouroboros.Consensus.Util.IOLike.yield
       pure pipeliningResult
 
     -- \| Clear a tentative header that turned out to be invalid. Also, roll
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB.hs
@@ -1,9 +1,7 @@
 {-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 module Ouroboros.Consensus.Storage.LedgerDB
   ( -- * API
@@ -17,11 +15,12 @@ module Ouroboros.Consensus.Storage.LedgerDB
   , openDBInternal
   ) where
 
-import Control.ResourceRegistry
 import Data.Functor.Contravariant ((>$<))
 import Data.Word
 import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.Config
 import Ouroboros.Consensus.HardFork.Abstract
+import Ouroboros.Consensus.Ledger.Extended
 import Ouroboros.Consensus.Ledger.Inspect
 import Ouroboros.Consensus.Ledger.SupportsProtocol
 import Ouroboros.Consensus.Storage.ImmutableDB.Stream
@@ -33,9 +32,7 @@ import Ouroboros.Consensus.Storage.LedgerDB.TraceEvent
 import qualified Ouroboros.Consensus.Storage.LedgerDB.V1 as V1
 import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.Snapshots as V1
 import qualified Ouroboros.Consensus.Storage.LedgerDB.V2 as V2
-import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.Args as V2
-import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.InMemory as InMemory
-import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.LSM as LSM
+import Ouroboros.Consensus.Storage.LedgerDB.V2.Backend
 import Ouroboros.Consensus.Util.Args
 import Ouroboros.Consensus.Util.CallStack
 import Ouroboros.Consensus.Util.IOLike
@@ -70,46 +67,35 @@ openDB
   stream
   replayGoal
   getBlock
-  getVolatileSuffix = case lgrFlavorArgs args of
-    LedgerDbFlavorArgsV1 bss ->
-      let snapManager = V1.snapshotManager args
-          initDb =
-            V1.mkInitDb
-              args
-              bss
-              getBlock
-              snapManager
-              getVolatileSuffix
-       in doOpenDB args initDb snapManager stream replayGoal
-    LedgerDbFlavorArgsV2 bss -> do
-      (snapManager, bss') <- case bss of
-        V2.V2Args V2.InMemoryHandleArgs -> pure (InMemory.snapshotManager args, V2.InMemoryHandleEnv)
-        V2.V2Args (V2.LSMHandleArgs (V2.LSMArgs path salt mkFS)) -> do
-          (rk1, V2.SomeHasFSAndBlockIO fs blockio) <- mkFS (lgrRegistry args)
-          session <-
-            allocate
-              (lgrRegistry args)
-              ( \_ ->
-                  LSM.openSession
-                    (LedgerDBFlavorImplEvent . FlavorImplSpecificTraceV2 . V2.LSMTrace >$< lgrTracer args)
-                    fs
-                    blockio
-                    salt
-                    path
-              )
-              LSM.closeSession
-          pure
-            ( LSM.snapshotManager (snd session) args
-            , V2.LSMHandleEnv (V2.LSMResources (fst session) (snd session) rk1)
-            )
-      let initDb =
-            V2.mkInitDb
-              args
-              bss'
-              getBlock
-              snapManager
-              getVolatileSuffix
-      doOpenDB args initDb snapManager stream replayGoal
+  getVolatileSuffix =
+    case lgrBackendArgs args of
+      LedgerDbBackendArgsV1 bss ->
+        let snapManager = V1.snapshotManager args
+            initDb =
+              V1.mkInitDb
+                args
+                bss
+                getBlock
+                snapManager
+                getVolatileSuffix
+         in doOpenDB args initDb snapManager stream replayGoal
+      LedgerDbBackendArgsV2 (SomeBackendArgs bArgs) -> do
+        res <-
+          mkResources
+            (Proxy @blk)
+            (LedgerDBFlavorImplEvent . FlavorImplSpecificTraceV2 >$< lgrTracer args)
+            bArgs
+            (lgrRegistry args)
+            (lgrHasFS args)
+        let snapManager =
+              snapshotManager
+                (Proxy @blk)
+                res
+                (configCodec . getExtLedgerCfg . ledgerDbCfg $ lgrConfig args)
+                (LedgerDBSnapshotEvent >$< lgrTracer args)
+                (lgrHasFS args)
+        let initDb = V2.mkInitDb args getBlock snapManager getVolatileSuffix res
+        doOpenDB args initDb snapManager stream replayGoal
 
 {-------------------------------------------------------------------------------
   Opening a LedgerDB
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/API.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/API.hs
@@ -6,12 +6,8 @@
 {-# LANGUAGE DerivingVia #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE PolyKinds #-}
-{-# LANGUAGE QuantifiedConstraints #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
@@ -154,22 +150,32 @@ module Ouroboros.Consensus.Storage.LedgerDB.API
     -- * Snapshots
   , SnapCounters (..)
 
+    -- * Streaming
+  , StreamingBackend (..)
+  , Yield
+  , Sink
+  , Decoders (..)
+
     -- * Testing
   , TestInternals (..)
   , TestInternals'
   , WhereToTakeSnapshot (..)
   ) where
 
+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
+import Data.ByteString (ByteString)
 import Data.Functor.Contravariant ((>$<))
 import Data.Kind
 import qualified Data.Map.Strict as Map
 import Data.MemPack
+import Data.Proxy
 import Data.Set (Set)
 import Data.Void (absurd)
 import Data.Word
@@ -195,6 +201,8 @@ import Ouroboros.Consensus.Util.IOLike
 import Ouroboros.Consensus.Util.IndexedMemPack
 import Ouroboros.Network.Block
 import Ouroboros.Network.Protocol.LocalStateQuery.Type
+import Streaming
+import System.FS.CRC
 
 {-------------------------------------------------------------------------------
   Main API
@@ -797,3 +805,41 @@ data LedgerDbPrune
     -- slot.
     LedgerDbPruneBeforeSlot SlotNo
   deriving Show
+
+{-------------------------------------------------------------------------------
+  Streaming
+-------------------------------------------------------------------------------}
+
+-- | A backend that supports streaming the ledger tables
+class StreamingBackend m backend l where
+  data YieldArgs m backend l
+
+  data SinkArgs m backend l
+
+  yield :: Proxy backend -> YieldArgs m backend l -> Yield m l
+
+  sink :: Proxy backend -> SinkArgs m backend l -> Sink m l
+
+type Yield m l =
+  l EmptyMK ->
+  ( ( Stream
+        (Of (TxIn l, TxOut l))
+        (ExceptT DeserialiseFailure m)
+        (Stream (Of ByteString) m (Maybe CRC)) ->
+      ExceptT DeserialiseFailure m (Stream (Of ByteString) m (Maybe CRC, Maybe CRC))
+    )
+  ) ->
+  ExceptT DeserialiseFailure m (Maybe CRC, Maybe CRC)
+
+type Sink m l =
+  l EmptyMK ->
+  Stream
+    (Of (TxIn l, TxOut l))
+    (ExceptT DeserialiseFailure m)
+    (Stream (Of ByteString) m (Maybe CRC)) ->
+  ExceptT DeserialiseFailure m (Stream (Of ByteString) m (Maybe CRC, Maybe CRC))
+
+data Decoders l
+  = Decoders
+      (forall s. Decoder s (TxIn l))
+      (forall s. Decoder s (TxOut l))
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/Args.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/Args.hs
@@ -1,25 +1,16 @@
-{-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE DerivingVia #-}
 {-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE GADTs #-}
 {-# LANGUAGE NumericUnderscores #-}
-{-# LANGUAGE PolyKinds #-}
-{-# LANGUAGE QuantifiedConstraints #-}
 {-# LANGUAGE RankNTypes #-}
-{-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneKindSignatures #-}
-{-# LANGUAGE TypeFamilies #-}
-{-# LANGUAGE UndecidableInstances #-}
 
 -- | Arguments for LedgerDB initialization.
 module Ouroboros.Consensus.Storage.LedgerDB.Args
   ( LedgerDbArgs (..)
   , LedgerDbBackendArgs (..)
-  , LedgerDbFlavorArgs (..)
   , QueryBatchSize (..)
   , defaultArgs
   , defaultQueryBatchSize
@@ -44,21 +35,13 @@ import Ouroboros.Consensus.Storage.LedgerDB.API
 import Ouroboros.Consensus.Storage.LedgerDB.Snapshots
 import Ouroboros.Consensus.Storage.LedgerDB.TraceEvent
 import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.Args as V1
-import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.Args as V2
+import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.Backend as V2
 import Ouroboros.Consensus.Util.Args
 import Ouroboros.Consensus.Util.IOLike
 import Ouroboros.Network.AnchoredSeq (AnchoredSeq)
 import qualified Ouroboros.Network.AnchoredSeq as AS
 import System.FS.API
 
-data LedgerDbBackendArgs m
-  = V1LMDB (Complete V1.LedgerDbFlavorArgs m)
-  | V2InMemory
-  | V2LSM
-      -- | The filepath **relative to the fast storage device** in which we will
-      -- open/create the LSM-tree database.
-      FilePath
-
 {-------------------------------------------------------------------------------
   Arguments
 -------------------------------------------------------------------------------}
@@ -75,7 +58,7 @@ data LedgerDbArgs f m blk = LedgerDbArgs
   , lgrHasFS :: HKD f (SomeHasFS m)
   , lgrConfig :: LedgerDbCfgF f (ExtLedgerState blk)
   , lgrTracer :: Tracer m (TraceEvent blk)
-  , lgrFlavorArgs :: LedgerDbFlavorArgs f m
+  , lgrBackendArgs :: LedgerDbBackendArgs m blk
   , lgrRegistry :: HKD f (ResourceRegistry m)
   , lgrQueryBatchSize :: QueryBatchSize
   , lgrStartSnapshot :: Maybe DiskSnapshot
@@ -87,8 +70,9 @@ data LedgerDbArgs f m blk = LedgerDbArgs
 -- | Default arguments
 defaultArgs ::
   Applicative m =>
+  V2.SomeBackendArgs m blk ->
   Incomplete LedgerDbArgs m blk
-defaultArgs =
+defaultArgs backendArgs =
   LedgerDbArgs
     { lgrSnapshotPolicyArgs = defaultSnapshotPolicyArgs
     , lgrGenesis = NoDefault
@@ -98,24 +82,24 @@ defaultArgs =
     , lgrTracer = nullTracer
     , -- This value is the closest thing to a pre-UTxO-HD node, and as such it
       -- will be the default for end-users.
-      lgrFlavorArgs = LedgerDbFlavorArgsV2 (V2.V2Args V2.InMemoryHandleArgs)
+      lgrBackendArgs = LedgerDbBackendArgsV2 backendArgs
     , lgrRegistry = NoDefault
     , lgrStartSnapshot = Nothing
     }
 
-data LedgerDbFlavorArgs f m
-  = LedgerDbFlavorArgsV1 (V1.LedgerDbFlavorArgs f m)
-  | LedgerDbFlavorArgsV2 (V2.LedgerDbFlavorArgs f m)
+data LedgerDbBackendArgs m blk
+  = LedgerDbBackendArgsV1 (V1.LedgerDbBackendArgs m (ExtLedgerState blk))
+  | LedgerDbBackendArgsV2 (V2.SomeBackendArgs m blk)
 
 {-------------------------------------------------------------------------------
   QueryBatchSize
 -------------------------------------------------------------------------------}
 
--- | The /maximum/ number of keys to read in a backing store range query.
+-- | The /maximum/ number of keys to read in a forker range query.
 --
 -- When performing a ledger state query that involves on-disk parts of the
 -- ledger state, we might have to read ranges of key-value pair data (e.g.,
--- UTxO) from disk using backing store range queries. Instead of reading all
+-- UTxO) from disk using forker range queries. Instead of reading all
 -- data in one go, we read it in batches. 'QueryBatchSize' determines the size
 -- of these batches.
 --
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/TraceEvent.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/LedgerDB/TraceEvent.hs
diff --git a/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Util/StreamingLedgerTables.hs b/ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Util/StreamingLedgerTables.hs
