content: {-# LANGUAGE DataKinds #-}
  {-# LANGUAGE DeriveAnyClass #-}
  {-# LANGUAGE DeriveGeneric #-}
  {-# LANGUAGE DerivingVia #-}
  {-# LANGUAGE FlexibleContexts #-}
  {-# LANGUAGE FlexibleInstances #-}
  {-# LANGUAGE LambdaCase #-}
  {-# LANGUAGE MultiParamTypeClasses #-}
  {-# LANGUAGE NamedFieldPuns #-}
  {-# LANGUAGE PatternSynonyms #-}
  {-# LANGUAGE ScopedTypeVariables #-}
  {-# LANGUAGE TupleSections #-}
  {-# LANGUAGE TypeApplications #-}
  {-# LANGUAGE TypeData #-}
  {-# LANGUAGE TypeFamilies #-}
  {-# LANGUAGE TypeOperators #-}
  {-# LANGUAGE UndecidableInstances #-}
  
  -- | A 'BackingStore' implementation based on [LMDB](http://www.lmdb.tech/doc/).
  module Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB
    ( -- * Opening a database
      LMDB
    , Backend (..)
    , Args (LMDBBackingStoreArgs)
    , LMDBLimits (LMDBLimits, lmdbMapSize, lmdbMaxDatabases, lmdbMaxReaders)
    , mkLMDBArgs
  
      -- * Streaming
    , YieldArgs (YieldLMDB)
    , mkLMDBYieldArgs
    , SinkArgs (SinkLMDB)
    , mkLMDBSinkArgs
  
      -- * Exposed for testing
    , LMDBErr (..)
    ) where
  
  import Cardano.Slotting.Slot (WithOrigin (At))
  import qualified Codec.Serialise as S (Serialise (..))
  import qualified Control.Concurrent.Class.MonadSTM.TVar as IOLike
  import Control.Monad (forM_, unless, void, when)
  import qualified Control.Monad.Class.MonadSTM as IOLike
  import Control.Monad.IO.Class (MonadIO (liftIO))
  import Control.Monad.Trans (lift)
  import Control.ResourceRegistry
  import qualified Control.Tracer as Trace
  import Data.Bifunctor (first)
  import Data.Functor (($>), (<&>))
  import Data.Functor.Contravariant ((>$<))
  import Data.Map (Map)
  import qualified Data.Map.Strict as Map
  import Data.MemPack
  import Data.Proxy
  import qualified Data.SOP.Dict as Dict
  import qualified Data.Set as Set
  import qualified Data.Text as Strict
  import qualified Database.LMDB.Simple as LMDB
  import qualified Database.LMDB.Simple.Cursor as LMDB.Cursor
  import qualified Database.LMDB.Simple.Extra as LMDB
  import qualified Database.LMDB.Simple.Internal as LMDB.Internal
  import qualified Database.LMDB.Simple.TransactionHandle as TrH
  import GHC.Generics (Generic)
  import GHC.Stack (HasCallStack)
  import Ouroboros.Consensus.Block
  import Ouroboros.Consensus.Ledger.Basics
  import qualified Ouroboros.Consensus.Ledger.Tables.Diff as Diff
  import Ouroboros.Consensus.Ledger.Tables.Utils (emptyLedgerTables)
  import Ouroboros.Consensus.Storage.LedgerDB.API
  import Ouroboros.Consensus.Storage.LedgerDB.Args
  import Ouroboros.Consensus.Storage.LedgerDB.Snapshots
    ( SnapshotBackend (..)
    )
  import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.Args as V1
  import Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore
  import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.API as API
  import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Bridge as Bridge
  import Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Status
    ( Status (..)
    , StatusLock
    )
  import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Status as Status
  import Ouroboros.Consensus.Util (foldlM')
  import Ouroboros.Consensus.Util.IOLike
    ( Exception (..)
    , IOLike
    , MonadCatch (..)
    , MonadThrow (..)
    , PrimState
    , bracket
    )
  import Ouroboros.Consensus.Util.IndexedMemPack
  import qualified Streaming as S
  import qualified Streaming.Prelude as S
  import System.Directory
  import qualified System.FS.API as FS
  import System.FS.IO
  import qualified System.FilePath as FilePath
  import System.IO.Temp
  
  {-------------------------------------------------------------------------------
    Database definition
  -------------------------------------------------------------------------------}
  
  -- | The LMDB database that underlies the backing store.
  data Db m l = Db
    { dbEnv :: !(LMDB.Environment LMDB.ReadWrite)
    -- ^ The LMDB environment is a pointer to the directory that contains the
    -- @`Db`@.
    , dbState :: !(LMDB.Database () DbSeqNo)
    -- ^ The on-disk state of the @`Db`@.
    --
    -- The state is kept in an LDMB table with only one key and one value:
    -- The current sequence number of the @`Db`@.
    , dbBackingTables :: !(LedgerTables l LMDBMK)
    -- ^ The LMDB tables with the key-value stores.
    , dbFilePath :: !FilePath
    , dbTracer :: !(Trace.Tracer m API.BackingStoreTrace)
    , dbStatusLock :: !(StatusLock m)
    -- ^ Status of the LMDB backing store. When 'Closed', all backing store
    -- (value handle) operations will fail.
    , dbOpenHandles :: !(IOLike.TVar m (Map Int (Cleanup m)))
    -- ^ Map of open value handles to cleanup actions. When closing the backing
    -- store, these cleanup actions are used to ensure all value handles cleaned
    -- up.
    --
    -- Note: why not use 'bsvhClose' here? We would get nested lock acquisition
    -- on 'dbStatusLock', which causes a deadlock:
    --
    -- * 'bsClose' acquires a write lock
    --
    -- * 'bsvhClose' is called on a value handle
    --
    -- * 'bsvhClose' tries to acquire a read lock, but it has to wait for
    --   'bsClose' to give up its write lock
    , dbNextId :: !(IOLike.TVar m Int)
    }
  
  newtype LMDBLimits = MkLMDBLimits {unLMDBLimits :: LMDB.Limits}
    deriving (Show, Eq)
  
  {-# COMPLETE LMDBLimits #-}
  
  -- | Configuration to use for LMDB backing store initialisation.
  --
  -- Keep the following in mind:
  --
  -- * @'lmdbMapSize'@ should be a multiple of the OS page size.
  --
  -- * @'lmdbMaxDatabases'@ should be set to at least 2, since the backing store
  --    has 2 internal LMDB databases by default: 1 for the actual tables, and
  --    1 for the database state @'DbSeqNo'@.
  pattern LMDBLimits :: Int -> Int -> Int -> LMDBLimits
  pattern LMDBLimits{lmdbMapSize, lmdbMaxDatabases, lmdbMaxReaders} =
    MkLMDBLimits
      LMDB.Limits
        { LMDB.mapSize = lmdbMapSize
        , LMDB.maxDatabases = lmdbMaxDatabases
        , LMDB.maxReaders = lmdbMaxReaders
        }
  
  -- | The database state consists of only the database sequence number @dbsSeq@.
  -- @dbsSeq@ represents the slot up to which we have flushed changes to disk.
  -- Note that we only flush changes to disk if they have become immutable.
  newtype DbSeqNo = DbSeqNo
    { dbsSeq :: WithOrigin SlotNo
    }
    deriving stock (Show, Generic)
    deriving anyclass S.Serialise
  
  -- | A 'MapKind' that represents an LMDB database handle
  data LMDBMK k v = LMDBMK !String !(LMDB.Database k v)
  
  {-------------------------------------------------------------------------------
    Low-level API
  -------------------------------------------------------------------------------}
  
  getDb ::
    LMDB.Internal.IsMode mode =>
    K2 String k v ->
    LMDB.Transaction mode (LMDBMK k v)
  getDb (K2 name) = LMDBMK name <$> LMDB.getDatabase (Just name)
  
  readAll ::
    (Ord (TxIn l), MemPack (TxIn l), IndexedMemPack idx (TxOut l)) =>
    Proxy l ->
    idx ->
    LMDBMK (TxIn l) (TxOut l) ->
    LMDB.Transaction mode (ValuesMK (TxIn l) (TxOut l))
  readAll _ st (LMDBMK _ dbMK) =
    ValuesMK
      <$> Bridge.runCursorAsTransaction'
        st
        LMDB.Cursor.cgetAll
        dbMK
  
  -- | @'rangeRead' rq dbMK@ performs a range read of @rqCount rq@
  -- values from database @dbMK@, starting from some key depending on @rqPrev rq@.
  --
  -- The @codec@ argument defines how to serialise/deserialise keys and values.
  --
  -- A range read can return less than @count@ values if there are not enough
  -- values to read.
  --
  -- What the "first" key in the database is, and more generally in which order
  -- keys are read, depends on the lexographical ordering of the /serialised/
  -- keys. Care should be taken such that the @'Ord'@ instance for @k@ matches the
  -- lexicographical ordering of the serialised keys, or the result of this
  -- function will be unexpected.
  rangeRead ::
    forall mode l idx.
    (Ord (TxIn l), MemPack (TxIn l), IndexedMemPack idx (TxOut l)) =>
    API.RangeQuery (LedgerTables l KeysMK) ->
    idx ->
    LMDBMK (TxIn l) (TxOut l) ->
    LMDB.Transaction mode (ValuesMK (TxIn l) (TxOut l), Maybe (TxIn l))
  rangeRead rq st dbMK =
    first ValuesMK <$> case ksMK of
      Nothing -> runCursorHelper Nothing
      Just (LedgerTables (KeysMK ks)) -> case Set.lookupMax ks of
        Nothing -> pure (mempty, Nothing)
        Just lastExcludedKey ->
          runCursorHelper $ Just (lastExcludedKey, LMDB.Cursor.Exclusive)
   where
    LMDBMK _ db = dbMK
  
    API.RangeQuery ksMK count = rq
  
    runCursorHelper ::
      Maybe (TxIn l, LMDB.Cursor.Bound) ->
      -- \^ Lower bound on read range
      LMDB.Transaction mode (Map (TxIn l) (TxOut l), Maybe (TxIn l))
    runCursorHelper lb =
      Bridge.runCursorAsTransaction'
        st
        (LMDB.Cursor.cgetManyAndLast lb count)
        db
  
  initLMDBTable ::
    (IndexedMemPack idx v, MemPack k) =>
    idx ->
    LMDBMK k v ->
    ValuesMK k v ->
    LMDB.Transaction LMDB.ReadWrite (EmptyMK k v)
  initLMDBTable st (LMDBMK tblName db) (ValuesMK utxoVals) =
    EmptyMK <$ lmdbInitTable
   where
    lmdbInitTable = do
      isEmpty <- LMDB.null db
      unless isEmpty $ liftIO . throwIO $ LMDBErrInitialisingNonEmpty tblName
      void $
        Map.traverseWithKey
          (Bridge.indexedPut st db)
          utxoVals
  
  readLMDBTable ::
    (IndexedMemPack idx v, MemPack k) =>
    Ord k =>
    idx ->
    LMDBMK k v ->
    KeysMK k v ->
    LMDB.Transaction mode (ValuesMK k v)
  readLMDBTable st (LMDBMK _ db) (KeysMK keys) =
    ValuesMK <$> lmdbReadTable
   where
    lmdbReadTable = foldlM' go Map.empty (Set.toList keys)
     where
      go m k =
        Bridge.indexedGet st db k <&> \case
          Nothing -> m
          Just v -> Map.insert k v m
  
  writeLMDBTable ::
    (IndexedMemPack idx v, MemPack k) =>
    idx ->
    LMDBMK k v ->
    DiffMK k v ->
    LMDB.Transaction LMDB.ReadWrite (EmptyMK k v)
  writeLMDBTable st (LMDBMK _ db) (DiffMK d) =
    EmptyMK <$ lmdbWriteTable
   where
    lmdbWriteTable = void $ Diff.traverseDeltaWithKey_ go d
     where
      go k de = case de of
        Diff.Delete -> void $ Bridge.delete db k
        Diff.Insert v -> Bridge.indexedPut st db k v
  
  {-------------------------------------------------------------------------------
   Db state
  -------------------------------------------------------------------------------}
  
  readDbSeqNoMaybeNull ::
    LMDB.Database () DbSeqNo ->
    LMDB.Transaction mode (Maybe DbSeqNo)
  readDbSeqNoMaybeNull db = LMDB.get db ()
  
  readDbSeqNo ::
    LMDB.Database () DbSeqNo ->
    LMDB.Transaction mode DbSeqNo
  readDbSeqNo db = readDbSeqNoMaybeNull db >>= maybe (liftIO . throwIO $ LMDBErrNoDbSeqNo) pure
  
  withDbSeqNoRW ::
    LMDB.Database () DbSeqNo ->
    (DbSeqNo -> LMDB.Transaction LMDB.ReadWrite (a, DbSeqNo)) ->
    LMDB.Transaction LMDB.ReadWrite a
  withDbSeqNoRW db f = withDbSeqNoRWMaybeNull db $ maybe (liftIO . throwIO $ LMDBErrNoDbSeqNo) f
  
  withDbSeqNoRWMaybeNull ::
    LMDB.Database () DbSeqNo ->
    (Maybe DbSeqNo -> LMDB.Transaction LMDB.ReadWrite (a, DbSeqNo)) ->
    LMDB.Transaction LMDB.ReadWrite a
  withDbSeqNoRWMaybeNull db f =
    readDbSeqNoMaybeNull db >>= f >>= \(r, sNew) -> LMDB.put db () (Just sNew) $> r
  
  {-------------------------------------------------------------------------------
   Guards
  -------------------------------------------------------------------------------}
  
  data GuardDbDir = DirMustExist | DirMustNotExist
  
  -- | Guard for the existence/non-existence of a database directory,
  -- and create it if missing.
  checkAndOpenDbDir ::
    (MonadIO m, IOLike m) =>
    GuardDbDir ->
    FS.SomeHasFS m ->
    FS.FsPath ->
    m FilePath
  checkAndOpenDbDir mustExistDir (FS.SomeHasFS fs) path = do
    fileEx <- FS.doesFileExist fs path
    when fileEx $
      throwIO $
        LMDBErrNotADir path
    dirEx <- FS.doesDirectoryExist fs path
    lmdbFileExists <-
      FS.doesFileExist fs path{FS.fsPathToList = FS.fsPathToList path ++ [Strict.pack "data.mdb"]}
    filepath <- FS.unsafeToFilePath fs path
    case dirEx of
      True
        | DirMustNotExist <- mustExistDir -> throwIO $ LMDBErrDirExists filepath
        | not lmdbFileExists -> throwIO $ LMDBErrDirIsNotLMDB filepath
        | otherwise -> pure ()
      False
        | DirMustExist <- mustExistDir -> throwIO $ LMDBErrDirDoesntExist filepath
        | otherwise -> pure ()
    FS.createDirectoryIfMissing fs True path
    pure filepath
  
  -- | Same as @`checkAndOpenDbDir`@, but retries the guard if we can make meaningful
  -- changes to the filesystem before we perform the retry.
  --
  -- Note: We only retry if a database directory exists while it shoudn't. In
  -- this case, we remove the directory recursively before retrying the guard.
  -- This is necessary for initialisation of the LMDB backing store, since the
  -- (non-snapshot) tables will probably still be on-disk. These tables are not
  -- removed when stopping the node, so they should be "overwritten".
  checkAndOpenDbDirWithRetry ::
    (MonadIO m, IOLike m) =>
    GuardDbDir ->
    FS.SomeHasFS m ->
    FS.FsPath ->
    m FilePath
  checkAndOpenDbDirWithRetry gdd shfs@(FS.SomeHasFS fs) path =
    handle retryHandler (checkAndOpenDbDir gdd shfs path)
   where
    retryHandler e = case (gdd, e) of
      (DirMustNotExist, LMDBErrDirExists _path) -> do
        FS.removeDirectoryRecursive fs path
        checkAndOpenDbDir DirMustNotExist shfs path
      _ -> throwIO e
  
  {-------------------------------------------------------------------------------
   Initialize an LMDB
  -------------------------------------------------------------------------------}
  
  -- | Initialise an LMDB database from these provided values.
  initFromVals ::
    forall l m.
    (HasLedgerTables l, MonadIO m, MemPackIdx l EmptyMK ~ l EmptyMK) =>
    Trace.Tracer m API.BackingStoreTrace ->
    -- | The slot number up to which the ledger tables contain values.
    WithOrigin SlotNo ->
    -- | The ledger tables to initialise the LMDB database tables with.
    LedgerTables l ValuesMK ->
    -- | The LMDB environment.
    LMDB.Environment LMDB.Internal.ReadWrite ->
    LMDB.Database () DbSeqNo ->
    l EmptyMK ->
    LedgerTables l LMDBMK ->
    m ()
  initFromVals tracer dbsSeq vals env st lst backingTables = do
    Trace.traceWith tracer $ API.BSInitialisingFromValues dbsSeq
    liftIO $
      LMDB.readWriteTransaction env $
        withDbSeqNoRWMaybeNull st $ \case
          Nothing ->
            ltzipWith2A (initLMDBTable lst) backingTables vals
              $> ((), DbSeqNo{dbsSeq})
          Just _ -> liftIO . throwIO $ LMDBErrInitialisingAlreadyHasState
    Trace.traceWith tracer $ API.BSInitialisedFromValues dbsSeq
  
  -- | Initialise an LMDB database from an existing LMDB database.
  initFromLMDBs ::
    (MonadIO m, IOLike m) =>
    Trace.Tracer m API.BackingStoreTrace ->
    -- | Configuration for the LMDB database that we initialise from.
    LMDBLimits ->
    -- | Abstraction over the filesystem.
    API.SnapshotsFS m ->
    -- | The path that contains the LMDB database that we want to initialise from.
    FS.FsPath ->
    -- | Abstraction over the filesystem.
    API.LiveLMDBFS m ->
    -- | The path where the new LMDB database should be initialised.
    FS.FsPath ->
    m ()
  initFromLMDBs tracer limits (API.SnapshotsFS shfsFrom@(FS.SomeHasFS fsFrom)) from0 (API.LiveLMDBFS shfsTo) to0 = do
    Trace.traceWith tracer $ API.BSInitialisingFromCopy from0
    from <- checkAndOpenDbDir DirMustExist shfsFrom from0
    -- On Windows, if we don't choose the mapsize carefully it will make the
    -- snapshot grow. Therefore we are using the current filesize as mapsize
    -- when opening the snapshot to avoid this.
    stat <-
      FS.withFile
        fsFrom
        (from0{FS.fsPathToList = FS.fsPathToList from0 ++ [Strict.pack "data.mdb"]})
        FS.ReadMode
        (FS.hGetSize fsFrom)
    to <- checkAndOpenDbDirWithRetry DirMustNotExist shfsTo to0
    bracket
      (liftIO $ LMDB.openEnvironment from ((unLMDBLimits limits){LMDB.mapSize = fromIntegral stat}))
      (liftIO . LMDB.closeEnvironment)
      (flip (lmdbCopy from0 tracer) to)
    Trace.traceWith tracer $ API.BSInitialisedFromCopy from0
  
  -- | Copy an existing LMDB database to a given directory.
  lmdbCopy ::
    MonadIO m =>
    FS.FsPath ->
    Trace.Tracer m API.BackingStoreTrace ->
    -- | The environment in which the LMDB database lives.
    LMDB.Environment LMDB.ReadWrite ->
    -- | The path where the copy should reside.
    FilePath ->
    m ()
  lmdbCopy from0 tracer e to = do
    Trace.traceWith tracer $ API.BSCopying from0
    liftIO $ LMDB.copyEnvironment e to
    Trace.traceWith tracer $ API.BSCopied from0
  
  -- | Initialise a backing store.
  newLMDBBackingStore ::
    forall m l.
    ( HasCallStack
    , HasLedgerTables l
    , MonadIO m
    , IOLike m
    , MemPackIdx l EmptyMK ~ l EmptyMK
    ) =>
    Trace.Tracer m API.BackingStoreTrace ->
    -- | Configuration parameters for the LMDB database that we
    -- initialise. In case we initialise the LMDB database from
    -- an existing LMDB database, we use these same configuration parameters
    -- to open the existing LMDB database.
    LMDBLimits ->
    -- | The FS for the LMDB live database
    API.LiveLMDBFS m ->
    API.SnapshotsFS m ->
    API.InitFrom (LedgerTables l ValuesMK) ->
    m (API.LedgerBackingStore m l)
  newLMDBBackingStore dbTracer limits liveFS@(API.LiveLMDBFS liveFS') snapFS@(API.SnapshotsFS snapFS') initFrom = do
    Trace.traceWith dbTracer API.BSOpening
  
    db@Db
      { dbEnv
      , dbState
      , dbBackingTables
      } <-
      createOrGetDB
  
    maybePopulate dbEnv dbState dbBackingTables
  
    Trace.traceWith dbTracer $ API.BSOpened $ Just path
  
    pure $ mkBackingStore db
   where
    path = FS.mkFsPath ["tables"]
  
    st = case initFrom of
      API.InitFromCopy st' _ -> st'
      API.InitFromValues _ st' _ -> st'
  
    createOrGetDB :: m (Db m l)
    createOrGetDB = do
      dbOpenHandles <- IOLike.newTVarIO Map.empty
      dbStatusLock <- Status.new Open
  
      -- get the filepath for this db creates the directory if appropriate
      dbFilePath <- checkAndOpenDbDirWithRetry DirMustNotExist liveFS' path
  
      -- copy from another lmdb path if appropriate
      case initFrom of
        API.InitFromCopy _ fp -> initFromLMDBs dbTracer limits snapFS fp liveFS path
        API.InitFromValues{} -> pure ()
  
      -- open this database
      dbEnv <- liftIO $ LMDB.openEnvironment dbFilePath (unLMDBLimits limits)
  
      -- The LMDB.Database that holds the @`DbSeqNo`@ (i.e. sequence number)
      -- This transaction must be read-write because on initialisation it creates the database
      dbState <- liftIO $ LMDB.readWriteTransaction dbEnv $ LMDB.getDatabase (Just "_dbstate")
  
      -- Here we get the LMDB.Databases for the tables of the ledger state
      -- Must be read-write transaction because tables may need to be created
      dbBackingTables <-
        liftIO $
          LMDB.readWriteTransaction dbEnv $
            lttraverse getDb (ltpure $ K2 "utxo")
  
      dbNextId <- IOLike.newTVarIO 0
  
      pure $
        Db
          { dbEnv
          , dbState
          , dbBackingTables
          , dbFilePath
          , dbTracer
          , dbStatusLock
          , dbOpenHandles
          , dbNextId
          }
  
    maybePopulate ::
      LMDB.Internal.Environment LMDB.Internal.ReadWrite ->
      LMDB.Internal.Database () DbSeqNo ->
      LedgerTables l LMDBMK ->
      m ()
    maybePopulate dbEnv dbState dbBackingTables = do
      -- now initialise those tables if appropriate
      case initFrom of
        API.InitFromValues slot _ vals -> initFromVals dbTracer slot vals dbEnv dbState st dbBackingTables
        API.InitFromCopy{} -> pure ()
  
    mkBackingStore :: HasCallStack => Db m l -> API.LedgerBackingStore m l
    mkBackingStore db =
      let bsClose :: m ()
          bsClose = Status.withWriteAccess dbStatusLock traceAlreadyClosed $ do
            Trace.traceWith dbTracer API.BSClosing
            openHandles <- IOLike.readTVarIO dbOpenHandles
            forM_ openHandles runCleanup
            IOLike.atomically $ IOLike.writeTVar dbOpenHandles mempty
            liftIO $ LMDB.closeEnvironment dbEnv
            Trace.traceWith dbTracer API.BSClosed
            pure ((), Closed)
           where
            traceAlreadyClosed = Trace.traceWith dbTracer API.BSAlreadyClosed
  
          bsCopy bsp = Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
            to <- checkAndOpenDbDir DirMustNotExist snapFS' bsp
            lmdbCopy path dbTracer dbEnv to
  
          bsValueHandle = Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
            mkLMDBBackingStoreValueHandle db
  
          bsWrite :: SlotNo -> (l EmptyMK, l EmptyMK) -> LedgerTables l DiffMK -> m ()
          bsWrite slot (_st, st') diffs = do
            Trace.traceWith dbTracer $ API.BSWriting slot
            Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
              oldSlot <- liftIO $ LMDB.readWriteTransaction dbEnv $ withDbSeqNoRW dbState $ \s@DbSeqNo{dbsSeq} -> do
                unless (dbsSeq <= At slot) $
                  -- This inequality is non-strict because of EBBs having the
                  -- same slot as its predecessor.
                  liftIO . throwIO $
                    LMDBErrNonMonotonicSeq (At slot) dbsSeq
                void $ ltzipWith2A (writeLMDBTable st') dbBackingTables diffs
                pure (dbsSeq, s{dbsSeq = At slot})
              Trace.traceWith dbTracer $ API.BSWritten oldSlot slot
       in API.BackingStore
            { API.bsClose = bsClose
            , API.bsCopy = const bsCopy
            , API.bsValueHandle = bsValueHandle
            , API.bsWrite = bsWrite
            , API.bsSnapshotBackend = UTxOHDLMDBSnapshot
            }
     where
      Db
        { dbEnv
        , dbState
        , dbBackingTables
        , dbStatusLock
        , dbOpenHandles
        } = db
  
  -- | Create a backing store value handle that has a consistent view of the
  -- current database state.
  mkLMDBBackingStoreValueHandle ::
    forall l m.
    (HasLedgerTables l, MonadIO m, IOLike m, HasCallStack, MemPackIdx l EmptyMK ~ l EmptyMK) =>
    -- | The LMDB database for which the backing store value handle is
    -- created.
    Db m l ->
    m (API.LedgerBackingStoreValueHandle m l)
  mkLMDBBackingStoreValueHandle db = do
    vhId <- IOLike.atomically $ do
      vhId <- IOLike.readTVar dbNextId
      IOLike.modifyTVar' dbNextId (+ 1)
      pure vhId
  
    let
      dbEnvRo = LMDB.readOnlyEnvironment dbEnv
      tracer = API.BSValueHandleTrace (Just vhId) >$< dbTracer
  
    Trace.traceWith dbTracer API.BSCreatingValueHandle
  
    trh <- liftIO $ TrH.newReadOnly dbEnvRo
    mbInitSlot <- liftIO $ TrH.submitReadOnly trh $ readDbSeqNoMaybeNull dbState
    initSlot <- liftIO $ maybe (throwIO LMDBErrUnableToReadSeqNo) (pure . dbsSeq) mbInitSlot
    vhStatusLock <- Status.new Open
  
    let
      -- \| Clean up a backing store value handle by committing its transaction
      -- handle.
      cleanup :: Cleanup m
      cleanup =
        Cleanup $
          liftIO $
            TrH.commit trh
  
      bsvhClose :: m ()
      bsvhClose =
        Status.withReadAccess dbStatusLock traceAlreadyClosed $ do
          Status.withWriteAccess vhStatusLock traceTVHAlreadyClosed $ do
            Trace.traceWith tracer API.BSVHClosing
            runCleanup cleanup
            IOLike.atomically $ IOLike.modifyTVar' dbOpenHandles (Map.delete vhId)
            Trace.traceWith tracer API.BSVHClosed
            pure ((), Closed)
       where
        traceAlreadyClosed = Trace.traceWith dbTracer API.BSAlreadyClosed
        traceTVHAlreadyClosed = Trace.traceWith tracer API.BSVHAlreadyClosed
  
      bsvhRead :: l EmptyMK -> LedgerTables l KeysMK -> m (LedgerTables l ValuesMK)
      bsvhRead st keys =
        Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
          Status.withReadAccess vhStatusLock (throwIO (LMDBErrNoValueHandle vhId)) $ do
            Trace.traceWith tracer API.BSVHReading
            res <-
              liftIO $
                TrH.submitReadOnly trh $
                  ltzipWith2A (readLMDBTable st) dbBackingTables keys
            Trace.traceWith tracer API.BSVHRead
            pure res
  
      bsvhRangeRead ::
        l EmptyMK ->
        API.RangeQuery (LedgerTables l KeysMK) ->
        m (LedgerTables l ValuesMK, Maybe (TxIn l))
      bsvhRangeRead st rq =
        Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
          Status.withReadAccess vhStatusLock (throwIO (LMDBErrNoValueHandle vhId)) $ do
            Trace.traceWith tracer API.BSVHRangeReading
            res <-
              liftIO $
                TrH.submitReadOnly trh $
                  let dbMK = getLedgerTables dbBackingTables
                   in first LedgerTables <$> rangeRead rq st dbMK
            Trace.traceWith tracer API.BSVHRangeRead
            pure res
  
      bsvhStat :: m API.Statistics
      bsvhStat =
        Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
          Status.withReadAccess vhStatusLock (throwIO (LMDBErrNoValueHandle vhId)) $ do
            Trace.traceWith tracer API.BSVHStatting
            let transaction = do
                  DbSeqNo{dbsSeq} <- readDbSeqNo dbState
                  constn <- lttraverse (\(LMDBMK _ dbx) -> K2 <$> LMDB.size dbx) dbBackingTables
                  let n = ltcollapse constn
                  pure $ API.Statistics dbsSeq n
            res <- liftIO $ TrH.submitReadOnly trh transaction
            Trace.traceWith tracer API.BSVHStatted
            pure res
  
      bsvhReadAll :: l EmptyMK -> m (LedgerTables l ValuesMK)
      bsvhReadAll st =
        Status.withReadAccess dbStatusLock (throwIO LMDBErrClosed) $ do
          Status.withReadAccess vhStatusLock (throwIO (LMDBErrNoValueHandle vhId)) $ do
            Trace.traceWith tracer API.BSVHRangeReading
            res <-
              liftIO $
                TrH.submitReadOnly trh $
                  let dbMK = getLedgerTables dbBackingTables
                   in LedgerTables <$> readAll (Proxy @l) st dbMK
            Trace.traceWith tracer API.BSVHRangeRead
            pure res
  
      bsvh =
        API.BackingStoreValueHandle
          { API.bsvhAtSlot = initSlot
          , API.bsvhClose = bsvhClose
          , API.bsvhRead = bsvhRead
          , API.bsvhReadAll = bsvhReadAll
          , API.bsvhRangeRead = bsvhRangeRead
          , API.bsvhStat = bsvhStat
          }
  
    IOLike.atomically $ IOLike.modifyTVar' dbOpenHandles (Map.insert vhId cleanup)
  
    Trace.traceWith dbTracer API.BSCreatedValueHandle
    pure bsvh
   where
    Db
      { dbEnv
      , dbTracer
      , dbState
      , dbOpenHandles
      , dbBackingTables
      , dbNextId
      , dbStatusLock
      } = db
  
  -- | A monadic action used for cleaning up resources.
  newtype Cleanup m = Cleanup {runCleanup :: m ()}
  
  {-------------------------------------------------------------------------------
   Errors
  -------------------------------------------------------------------------------}
  
  -- | Errors that can be thrown by LMDB.
  --
  -- __WARNING__: these errors will be thrown in IO as having a corrupt database
  -- is critical for the functioning of Consensus.
  data LMDBErr
    = -- | The database state can not be found on-disk.
      LMDBErrNoDbSeqNo
    | -- | The sequence number of a @`Db`@ should be monotonically increasing
      -- across calls to @`bsWrite`@, since we use @`bsWrite`@ to flush
      -- /immutable/ changes. That is, we can only flush with a newer sequence
      -- number because the changes should be /immutable/. Note that this does
      -- not mean that values can not be changed in the future, only that we
      -- can not change values in the past.
      LMDBErrNonMonotonicSeq !(WithOrigin SlotNo) !(WithOrigin SlotNo)
    | -- | The database table that is being initialised is non-empty.
      LMDBErrInitialisingNonEmpty !String
    | -- | The database that is being initialized already had a DbSeqNo table
      LMDBErrInitialisingAlreadyHasState
    | -- | Trying to use a non-existing value handle.
      LMDBErrNoValueHandle !Int
    | -- | Couldn't create a value handle because we couldn't read the sequence
      -- number
      LMDBErrUnableToReadSeqNo
    | -- | Failed to read a value from a database table.
      LMDBErrBadRead
    | -- | Failed to read a range of values from a database table.
      LMDBErrBadRangeRead
    | -- | A database directory should not exist already.
      LMDBErrDirExists !FilePath
    | -- | A database directory should exist already.
      LMDBErrDirDoesntExist !FilePath
    | -- | The directory exists but is not an LMDB directory!
      LMDBErrDirIsNotLMDB !FilePath
    | -- | What should be a directory is in fact a file
      LMDBErrNotADir !FS.FsPath
    | -- | The database has been closed, so all backing store operations should
      -- throw an error.
      LMDBErrClosed
  
  instance Exception LMDBErr
  
  -- | Show instance for pretty printing @`LMDBErr`@s as error messages that
  -- include: (i) an indication of the probable cause of the error, and
  -- (ii) a descriptive error message for the specific @`LMDBErr`@.
  instance Show LMDBErr where
    show dbErr =
      mconcat
        [ "[LMDB-ERROR] "
        , "The LMDB Backing store has encountered a fatal exception. "
        , "Possibly, the LMDB database is corrupted.\n"
        , "[ERROR-MSG] "
        , prettyPrintLMDBErr dbErr
        ]
  
  -- | Pretty print a @`LMDBErr`@ with a descriptive error message.
  prettyPrintLMDBErr :: LMDBErr -> String
  prettyPrintLMDBErr = \case
    LMDBErrNoDbSeqNo ->
      "Can not find the database state on-disk."
    LMDBErrNonMonotonicSeq s1 s2 ->
      "Trying to write to the database with a non-monotonic sequence number: "
        <> showParen True (shows s1) ""
        <> " is not <= "
        <> showParen True (shows s2) ""
    LMDBErrInitialisingNonEmpty s ->
      "The database table that is being initialised is non-empty: " <> s
    LMDBErrInitialisingAlreadyHasState ->
      "The database contains no values but still has a table with a sequence number."
    LMDBErrNoValueHandle vh_id ->
      "Trying to use non-existing value handle: " <> show vh_id
    LMDBErrUnableToReadSeqNo ->
      "Reading the sequence number failed thus we couldn't create a value handle."
    LMDBErrBadRead ->
      "Failed to read a value from a database table."
    LMDBErrBadRangeRead ->
      "Failed to read a range of values from a database table."
    LMDBErrDirExists path ->
      "Database directory should not exist already: " <> show path
    LMDBErrDirDoesntExist path ->
      "Database directory should already exist: " <> show path
    LMDBErrDirIsNotLMDB path ->
      "Database directory doesn't contain an LMDB database: "
        <> show path
        <> "\nPre-UTxO-HD and In-Memory implementations are incompatible \
           \ with the LMDB implementation, please delete your ledger database \
           \ if you want to run with LMDB"
    LMDBErrNotADir path ->
      "The path " <> show path <> " should be a directory but it is a file instead."
    LMDBErrClosed -> "The database has been closed."
  
  {-------------------------------------------------------------------------------
    Backend
  -------------------------------------------------------------------------------}
  
  type data LMDB
  
  instance
    ( HasLedgerTables l
    , MonadIO m
    , IOLike m
    , MemPackIdx l EmptyMK ~ l EmptyMK
    ) =>
    Backend m LMDB l
    where
    data Args m LMDB
      = LMDBBackingStoreArgs FilePath LMDBLimits (Dict.Dict MonadIOPrim m)
    data Trace m LMDB
      = OnDiskBackingStoreInitialise LMDB.Limits
      | OnDiskBackingStoreTrace BackingStoreTrace
      deriving (Eq, Show)
  
    isRightBackendForSnapshot _ _ UTxOHDLMDBSnapshot = True
    isRightBackendForSnapshot _ _ _ = False
  
    newBackingStoreInitialiser trcr (LMDBBackingStoreArgs fs limits Dict.Dict) =
      newLMDBBackingStore
        (SomeBackendTrace . OnDiskBackingStoreTrace >$< trcr)
        limits
        (LiveLMDBFS $ FS.SomeHasFS $ ioHasFS $ FS.MountPoint fs)
  
  -- | Create arguments for initializing the LedgerDB using the LMDB backend.
  mkLMDBArgs ::
    ( MonadIOPrim m
    , HasLedgerTables (LedgerState blk)
    , IOLike m
    ) =>
    V1.FlushFrequency -> FilePath -> LMDBLimits -> a -> (LedgerDbBackendArgs m blk, a)
  mkLMDBArgs flushing lmdbPath limits =
    (,) $
      LedgerDbBackendArgsV1 $
        V1.V1Args flushing $
          SomeBackendArgs $
            LMDBBackingStoreArgs lmdbPath limits Dict.Dict
  
  class (MonadIO m, PrimState m ~ PrimState IO) => MonadIOPrim m
  instance (MonadIO m, PrimState m ~ PrimState IO) => MonadIOPrim m
  
  {-------------------------------------------------------------------------------
    Streaming
  -------------------------------------------------------------------------------}
  
  instance (Ord (TxIn l), GetTip l, Monad m) => StreamingBackend m LMDB l where
    data SinkArgs m LMDB l
      = SinkLMDB
          -- \| Chunk size
          Int
          -- \| bsWrite
          ( SlotNo ->
            (l EmptyMK, l EmptyMK) ->
            LedgerTables l DiffMK ->
            m ()
          )
          (l EmptyMK -> m ())
  
    data YieldArgs m LMDB l
      = YieldLMDB
          Int
          (LedgerBackingStoreValueHandle m l)
  
    yield _ (YieldLMDB chunkSize valueHandle) = yieldLmdbS chunkSize valueHandle
    sink _ (SinkLMDB chunkSize write copy) = sinkLmdbS chunkSize write copy
  
  sinkLmdbS ::
    forall m l.
    (Ord (TxIn l), GetTip l, Monad m) =>
    Int ->
    (SlotNo -> (l EmptyMK, l EmptyMK) -> LedgerTables l DiffMK -> m ()) ->
    (l EmptyMK -> m ()) ->
    Sink m l
  sinkLmdbS writeChunkSize bs copyTo hint s = do
    r <- go writeChunkSize mempty s
    lift $ copyTo hint
    pure (fmap (,Nothing) r)
   where
    sl = withOrigin (error "unreachable") id $ pointSlot $ getTip hint
  
    go 0 m s' = do
      lift $ bs sl (hint, hint) (LedgerTables $ DiffMK $ Diff.fromMapInserts m)
      go writeChunkSize mempty s'
    go n m s' = do
      mbs <- S.uncons s'
      case mbs of
        Nothing -> do
          lift $ bs sl (hint, hint) (LedgerTables $ DiffMK $ Diff.fromMapInserts m)
          S.effects s'
        Just ((k, v), s'') ->
          go (n - 1) (Map.insert k v m) s''
  
  yieldLmdbS ::
    Monad m =>
    Int ->
    LedgerBackingStoreValueHandle m l ->
    Yield m l
  yieldLmdbS readChunkSize bsvh hint k = do
    r <- k (go (RangeQuery Nothing readChunkSize))
    lift $ S.effects r
   where
    go p = do
      (LedgerTables (ValuesMK values), mx) <- lift $ S.lift $ bsvhRangeRead bsvh hint p
      case mx of
        Nothing -> pure $ pure Nothing
        Just x -> do
          S.each $ Map.toList values
          go (RangeQuery (Just . LedgerTables . KeysMK $ Set.singleton x) readChunkSize)
  
  -- | Create Yield args for LMDB
  mkLMDBYieldArgs ::
    forall l.
    ( HasCallStack
    , HasLedgerTables l
    , MemPackIdx l EmptyMK ~ l EmptyMK
    ) =>
    FilePath ->
    LMDBLimits ->
    l EmptyMK ->
    ResourceRegistry IO ->
    IO (YieldArgs IO LMDB l)
  mkLMDBYieldArgs fp limits hint reg = do
    let (dbPath, snapName) = FilePath.splitFileName fp
    tempDir <- getCanonicalTemporaryDirectory
    let lmdbTemp = tempDir FilePath.</> "lmdb_streaming_in"
    removePathForcibly lmdbTemp
    _ <-
      allocate
        reg
        (\_ -> createDirectory lmdbTemp)
        (\_ -> removePathForcibly lmdbTemp)
    (_, bs) <-
      allocate
        reg
        ( \_ -> do
            newLMDBBackingStore
              Trace.nullTracer
              limits
              (LiveLMDBFS $ FS.SomeHasFS $ ioHasFS $ FS.MountPoint lmdbTemp)
              (SnapshotsFS $ FS.SomeHasFS $ ioHasFS $ FS.MountPoint dbPath)
              (InitFromCopy hint (FS.mkFsPath [snapName]))
        )
        bsClose
    (_, bsvh) <- allocate reg (\_ -> bsValueHandle bs) bsvhClose
    pure (YieldLMDB 1000 bsvh)
  
  -- | Create Sink args for LMDB
  mkLMDBSinkArgs ::
    forall l.
    ( HasCallStack
    , HasLedgerTables l
    , MemPackIdx l EmptyMK ~ l EmptyMK
    ) =>
    FilePath ->
    LMDBLimits ->
    l EmptyMK ->
    ResourceRegistry IO ->
    IO (SinkArgs IO LMDB l)
  mkLMDBSinkArgs fp limits hint reg = do
    let (snapDir, snapName) = FilePath.splitFileName fp
    tempDir <- getCanonicalTemporaryDirectory
    let lmdbTemp = tempDir FilePath.</> "lmdb_streaming_out"
    removePathForcibly lmdbTemp
    _ <- allocate reg (\_ -> createDirectory lmdbTemp) (\_ -> removePathForcibly lmdbTemp)
    (_, bs) <-
      allocate
        reg
        ( \_ ->
            newLMDBBackingStore
              Trace.nullTracer
              limits
              (LiveLMDBFS $ FS.SomeHasFS $ ioHasFS $ FS.MountPoint lmdbTemp)
              (SnapshotsFS $ FS.SomeHasFS $ ioHasFS $ FS.MountPoint snapDir)
              (InitFromValues (At 0) hint emptyLedgerTables)
        )
        bsClose
    pure $ SinkLMDB 1000 (bsWrite bs) (\h -> bsCopy bs h (FS.mkFsPath [snapName, "tables"]))
  