Support LSM in snapshot-converter and streaming ledger tables

ouroboros-consensus-cardano/changelog.d/20250904_111240_javier.sagredo_lsm_sc_da.md

@@ -0,0 +1,25 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->
+<!--
+### Non-Breaking
+
+- A bullet item for the Non-Breaking category.
+
+-->
+<!--
+### Breaking
+
+- A bullet item for the Breaking category.
+
+-->

ouroboros-consensus-cardano/ouroboros-consensus-cardano.cabal

@@ -101,6 +101,7 @@ library
     Ouroboros.Consensus.Cardano.Ledger
     Ouroboros.Consensus.Cardano.Node
     Ouroboros.Consensus.Cardano.QueryHF
+    Ouroboros.Consensus.Cardano.StreamingLedgerTables
     Ouroboros.Consensus.Shelley.Crypto
     Ouroboros.Consensus.Shelley.Eras
     Ouroboros.Consensus.Shelley.HFEras
@@ -156,9 +157,13 @@ library
     cardano-strict-containers,
     cborg ^>=0.2.2,
     containers >=0.5 && <0.8,
+    contra-tracer,
     crypton,
     deepseq,
+    directory,
+    filepath,
     formatting >=6.3 && <7.3,
+    fs-api,
     measures,
     mempack,
     microlens,
@@ -167,12 +172,15 @@ library
     ouroboros-consensus ^>=0.27,
     ouroboros-consensus-protocol ^>=0.12,
     ouroboros-network-api ^>=0.16,
+    random,
+    resource-registry,
     serialise ^>=0.2,
     singletons ^>=3.0,
     small-steps,
     sop-core ^>=0.5,
     sop-extras ^>=0.4,
     strict-sop-core ^>=0.1,
+    temporary,
     text,
     these ^>=1.2,
     validation,
@@ -690,10 +698,10 @@ executable snapshot-converter
   hs-source-dirs: app
   main-is: snapshot-converter.hs
   build-depends:
+    ansi-terminal,
     base,
-    bytestring,
     cardano-crypto-class,
-    contra-tracer,
+    directory,
     filepath,
     fs-api,
     mtl,
@@ -703,8 +711,8 @@ executable snapshot-converter
     ouroboros-consensus-cardano:unstable-cardano-tools,
     resource-registry,
     serialise,
-    sop-core,
-    temporary,
+    terminal-progress-bar,
+    text,
     with-utf8,
 
   other-modules:

ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/StreamingLedgerTables.hs

@@ -0,0 +1,219 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+{-# LANGUAGE TypeOperators #-}
+
+module Ouroboros.Consensus.Cardano.StreamingLedgerTables
+  ( fromInMemory
+  , fromLSM
+  , fromLMDB
+  , toLMDB
+  , toLSM
+  , toInMemory
+  ) where
+
+import Cardano.Ledger.BaseTypes (WithOrigin (..))
+import Cardano.Ledger.Binary
+import Cardano.Ledger.Core (ByronEra, Era, eraDecoder, toEraCBOR)
+import qualified Cardano.Ledger.Shelley.API as SL
+import qualified Cardano.Ledger.Shelley.LedgerState as SL
+import qualified Cardano.Ledger.State as SL
+import qualified Codec.CBOR.Encoding
+import Control.ResourceRegistry
+import Control.Tracer (nullTracer)
+import Data.Proxy
+import Data.SOP.BasicFunctors
+import Data.SOP.Functors
+import Data.SOP.Strict
+import qualified Data.SOP.Telescope as Telescope
+import qualified Data.Text as T
+import Lens.Micro
+import Ouroboros.Consensus.Byron.Ledger
+import Ouroboros.Consensus.Cardano.Block
+import Ouroboros.Consensus.Cardano.Ledger
+import Ouroboros.Consensus.HardFork.Combinator
+import Ouroboros.Consensus.HardFork.Combinator.State
+import Ouroboros.Consensus.Ledger.Abstract
+import Ouroboros.Consensus.Ledger.Tables.Utils (emptyLedgerTables)
+import Ouroboros.Consensus.Shelley.Ledger
+import Ouroboros.Consensus.Shelley.Ledger.SupportsProtocol ()
+import Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.API
+import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB as LMDB
+import Ouroboros.Consensus.Storage.LedgerDB.V2.Args
+import Ouroboros.Consensus.Storage.LedgerDB.V2.LSM
+import Ouroboros.Consensus.Util.StreamingLedgerTables
+import System.Directory
+import System.FS.API
+import System.FS.IO
+import System.FilePath as FilePath
+import System.IO.Temp
+import System.Random
+
+type L = LedgerState (CardanoBlock StandardCrypto)
+
+fromInMemory :: FilePath -> L EmptyMK -> ResourceRegistry IO -> IO (YieldArgs L IO)
+fromInMemory fp (HardForkLedgerState (HardForkState idx)) _ =
+  let
+    np ::
+      NP
+        (Current (Flip LedgerState EmptyMK) -.-> K (Decoders L))
+        (CardanoEras StandardCrypto)
+    np =
+      (Fn $ const $ K $ error "Byron")
+        :* (Fn $ K . fromEra ShelleyTxOut . unFlip . currentState)
+        :* (Fn $ K . fromEra AllegraTxOut . unFlip . currentState)
+        :* (Fn $ K . fromEra MaryTxOut . unFlip . currentState)
+        :* (Fn $ K . fromEra AlonzoTxOut . unFlip . currentState)
+        :* (Fn $ K . fromEra BabbageTxOut . unFlip . currentState)
+        :* (Fn $ K . fromEra ConwayTxOut . unFlip . currentState)
+        :* (Fn $ K . fromEra DijkstraTxOut . unFlip . currentState)
+        :* Nil
+   in
+    pure $
+      YieldInMemory
+        (SomeHasFS . ioHasFS)
+        fp
+        (hcollapse $ hap np $ Telescope.tip idx)
+ where
+  fromEra ::
+    forall proto era.
+    ShelleyCompatible proto era =>
+    (TxOut (LedgerState (ShelleyBlock proto era)) -> CardanoTxOut StandardCrypto) ->
+    LedgerState (ShelleyBlock proto era) EmptyMK ->
+    Decoders L
+  fromEra toCardanoTxOut st =
+    let certInterns =
+          internsFromMap $
+            shelleyLedgerState st
+              ^. SL.nesEsL
+                . SL.esLStateL
+                . SL.lsCertStateL
+                . SL.certDStateL
+                . SL.accountsL
+                . SL.accountsMapL
+     in Decoders
+          (eraDecoder @era decodeMemPack)
+          (eraDecoder @era $ toCardanoTxOut <$> decShareCBOR certInterns)
+
+fromLMDB :: FilePath -> LMDB.LMDBLimits -> L EmptyMK -> ResourceRegistry IO -> IO (YieldArgs L IO)
+fromLMDB fp limits hint reg = do
+  let (dbPath, snapName) = splitFileName fp
+  tempDir <- getCanonicalTemporaryDirectory
+  let lmdbTemp = tempDir FilePath.</> "lmdb_streaming_in"
+  removePathForcibly lmdbTemp
+  _ <-
+    allocate
+      reg
+      (\_ -> System.Directory.createDirectory lmdbTemp)
+      (\_ -> removePathForcibly lmdbTemp)
+  (_, bs) <-
+    allocate
+      reg
+      ( \_ -> do
+          LMDB.newLMDBBackingStore
+            nullTracer
+            limits
+            (LiveLMDBFS $ SomeHasFS $ ioHasFS $ MountPoint lmdbTemp)
+            (SnapshotsFS $ SomeHasFS $ ioHasFS $ MountPoint dbPath)
+            (InitFromCopy hint (mkFsPath [snapName]))
+      )
+      bsClose
+  (_, bsvh) <- allocate reg (\_ -> bsValueHandle bs) bsvhClose
+  pure (YieldLMDB 1000 bsvh)
+
+fromLSM ::
+  FilePath ->
+  String ->
+  L EmptyMK ->
+  ResourceRegistry IO ->
+  IO (YieldArgs L IO)
+fromLSM fp snapName _ reg = do
+  (_, SomeHasFSAndBlockIO hasFS blockIO) <- stdMkBlockIOFS fp reg
+  salt <- fst . genWord64 <$> newStdGen
+  (_, session) <-
+    allocate reg (\_ -> openSession nullTracer hasFS blockIO salt (mkFsPath [])) closeSession
+  tb <-
+    allocate
+      reg
+      ( \_ ->
+          openTableFromSnapshot
+            session
+            (toSnapshotName snapName)
+            (SnapshotLabel $ T.pack "UTxO table")
+      )
+      closeTable
+  YieldLSM 1000 <$> newLSMLedgerTablesHandle nullTracer reg tb
+
+toLMDB ::
+  FilePath ->
+  LMDB.LMDBLimits ->
+  L EmptyMK ->
+  ResourceRegistry IO ->
+  IO (SinkArgs L IO)
+toLMDB fp limits hint reg = do
+  let (snapDir, snapName) = splitFileName fp
+  tempDir <- getCanonicalTemporaryDirectory
+  let lmdbTemp = tempDir FilePath.</> "lmdb_streaming_out"
+  removePathForcibly lmdbTemp
+  _ <-
+    allocate reg (\_ -> System.Directory.createDirectory lmdbTemp) (\_ -> removePathForcibly lmdbTemp)
+  (_, bs) <-
+    allocate
+      reg
+      ( \_ ->
+          LMDB.newLMDBBackingStore
+            nullTracer
+            limits
+            (LiveLMDBFS $ SomeHasFS $ ioHasFS $ MountPoint lmdbTemp)
+            (SnapshotsFS $ SomeHasFS $ ioHasFS $ MountPoint snapDir)
+            (InitFromValues (At 0) hint emptyLedgerTables)
+      )
+      bsClose
+  pure $ SinkLMDB 1000 (bsWrite bs) (\h -> bsCopy bs h (mkFsPath [snapName, "tables"]))
+
+toInMemory ::
+  FilePath ->
+  L EmptyMK ->
+  ResourceRegistry IO ->
+  IO (SinkArgs L IO)
+toInMemory fp (HardForkLedgerState (HardForkState idx)) _ = do
+  currDir <- getCurrentDirectory
+  let
+    np =
+      (Fn $ const $ K $ encOne (Proxy @ByronEra))
+        :* (Fn $ const $ K $ encOne (Proxy @ShelleyEra))
+        :* (Fn $ const $ K $ encOne (Proxy @AllegraEra))
+        :* (Fn $ const $ K $ encOne (Proxy @MaryEra))
+        :* (Fn $ const $ K $ encOne (Proxy @AlonzoEra))
+        :* (Fn $ const $ K $ encOne (Proxy @BabbageEra))
+        :* (Fn $ const $ K $ encOne (Proxy @ConwayEra))
+        :* (Fn $ const $ K $ encOne (Proxy @DijkstraEra))
+        :* Nil
+  pure $
+    uncurry
+      (SinkInMemory 1000)
+      (hcollapse $ hap np $ Telescope.tip idx)
+      (SomeHasFS $ ioHasFS $ MountPoint currDir)
+      fp
+ where
+  encOne ::
+    forall era.
+    Era era =>
+    Proxy era ->
+    (TxIn L -> Codec.CBOR.Encoding.Encoding, TxOut L -> Codec.CBOR.Encoding.Encoding)
+  encOne _ =
+    (toEraCBOR @era . encodeMemPack, toEraCBOR @era . eliminateCardanoTxOut (const encodeMemPack))
+
+toLSM ::
+  FilePath ->
+  String ->
+  L EmptyMK ->
+  ResourceRegistry IO ->
+  IO (SinkArgs L IO)
+toLSM fp snapName _ reg = do
+  removePathForcibly fp
+  System.Directory.createDirectory fp
+  (_, SomeHasFSAndBlockIO hasFS blockIO) <- stdMkBlockIOFS fp reg
+  salt <- fst . genWord64 <$> newStdGen
+  (_, session) <-
+    allocate reg (\_ -> newSession nullTracer hasFS blockIO salt (mkFsPath [])) closeSession
+  pure (SinkLSM 1000 snapName session)

ouroboros-consensus/changelog.d/20250904_111212_javier.sagredo_lsm_sc_da.md

@@ -0,0 +1,23 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+<!--
+### Patch
+
+- A bullet item for the Patch category.
+
+-->
+<!--
+### Non-Breaking
+
+- A bullet item for the Non-Breaking category.
+
+-->
+
+### Breaking
+
+- Removed the `readAll` functionality from the LedgerDB backends now that it became unnecessary as tables conversion is now streamed.

ouroboros-consensus/ouroboros-consensus.cabal

@@ -299,6 +299,7 @@ library
     Ouroboros.Consensus.Util.Orphans
     Ouroboros.Consensus.Util.RedundantConstraints
     Ouroboros.Consensus.Util.STM
+    Ouroboros.Consensus.Util.StreamingLedgerTables
     Ouroboros.Consensus.Util.Time
     Ouroboros.Consensus.Util.Versioned
 
@@ -326,6 +327,7 @@ library
     deepseq,
     diff-containers >=1.2,
     filelock,
+    filepath,
     fingertree-rm >=1.0,
     fs-api ^>=0.4,
     hashable,

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

@@ -54,7 +54,6 @@ module Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.API
 
     -- * 🧪 Testing
   , bsRead
-  , bsReadAll
   ) where
 
 import Cardano.Slotting.Slot (SlotNo, WithOrigin (..))
@@ -169,9 +168,6 @@ data BackingStoreValueHandle m keys key values = BackingStoreValueHandle
   -- itself is idempotent.
   , bsvhRangeRead :: !(ReadHint values -> RangeQuery keys -> m (values, Maybe key))
   -- ^ See 'RangeQuery'
-  , bsvhReadAll :: !(ReadHint values -> m values)
-  -- ^ Costly read all operation, not to be used in Consensus but only in
-  -- snapshot-converter executable.
   , bsvhRead :: !(ReadHint values -> keys -> m values)
   -- ^ Read the given keys from the handle
   --
@@ -206,7 +202,6 @@ castBackingStoreValueHandle f g h bsvh =
   BackingStoreValueHandle
     { bsvhAtSlot
     , bsvhClose
-    , bsvhReadAll = \rhint -> f <$> bsvhReadAll rhint
     , bsvhRangeRead = \rhint (RangeQuery prev count) ->
         fmap (second (fmap h) . first f) . bsvhRangeRead rhint $ RangeQuery (fmap g prev) count
     , bsvhRead = \rhint -> fmap f . bsvhRead rhint . g
@@ -215,7 +210,6 @@ castBackingStoreValueHandle f g h bsvh =
  where
   BackingStoreValueHandle
     { bsvhClose
-    , bsvhReadAll
     , bsvhAtSlot
     , bsvhRangeRead
     , bsvhRead
@@ -233,13 +227,6 @@ bsRead store rhint keys = withBsValueHandle store $ \vh -> do
   values <- bsvhRead vh rhint keys
   pure (bsvhAtSlot vh, values)
 
-bsReadAll ::
-  MonadThrow m =>
-  BackingStore m keys key values diff ->
-  ReadHint values ->
-  m values
-bsReadAll store rhint = withBsValueHandle store $ \vh -> bsvhReadAll vh rhint
-
 -- | A 'IOLike.bracket'ed 'bsValueHandle'
 withBsValueHandle ::
   MonadThrow m =>