{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}

-- | Maintenance of LeiosDB's volatile and immutable partitions.
--
--   * promotion: volatile EBs are copied to the immutable partition once they are old enough;
--   * garbage collection of the volatile partition;
--   * the in-memory 'LeiosDbStats'.
module Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Maintenance
  ( -- * Promotion to the immutable partition
    sqlPromoteToImmutable
  , startCopier

    -- * GC mark
  , sqlGarbageCollect
  , gcMark

    -- * GC sweep
  , SweeperConn (..)
  , sweepEbBatch

    -- * GC batch sizes
  , defaultGcBatchSize

    -- * Stats
  , initialStats
  , startVolatileStatsSampler
  , bumpVolatileStats
  , bumpVolatileStatsVar
  , bumpImmutableStats
  ) where

import Cardano.Slotting.Slot (SlotNo (..))
import Control.Concurrent (threadDelay)
import Control.Concurrent.Class.MonadSTM.Strict
  ( StrictTVar
  , check
  , modifyTVar
  , readTVar
  , readTVarIO
  , writeTVar
  )
import Control.Monad (filterM, forever, unless, void)
import Control.Monad.Class.MonadThrow
  ( bracket
  , catch
  , displayException
  )
import Control.ResourceRegistry
  ( ResourceRegistry
  , Thread
  , forkLinkedThread
  )
import Control.Tracer (Tracer, traceWith)
import Data.Int (Int64)
import Data.String (fromString)
import qualified Database.SQLite3.Direct as DB
import qualified GHC.Conc as IO (atomically)
import GHC.Stack (HasCallStack)
import Ouroboros.Consensus.Leios.Types
  ( EndorserBlockHash
  , LeiosPoint (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.API (Promise (..))
import Ouroboros.Consensus.Storage.LeiosDB.Exception
  ( LeiosDbException (..)
  , throwLeiosDbException
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Connection
  ( Conn (..)
  , closeChecked
  , openRawConnection
  , withReadOnlyConn
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Primitives
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Queries
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Statements
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.WriteQueue
  ( WriteJob (..)
  , WriteQueue
  , submitJob
  )
import Ouroboros.Consensus.Storage.LeiosDB.Trace (LeiosDbStats (..), TraceLeiosDb (..))
import Ouroboros.Consensus.Util.IOLike (atomically)
import System.Directory (doesFileExist, getFileSize)

-- * Copying EBs to the immutable partition

-- | Implements 'leiosDbPromoteToImmutable':
--   - synchronously pin the EB rows in the volatile partition for promotion (@status@ 0 -> 1),
--     through the writer ('PinEb');
--   - ring the copier's doorbell.
sqlPromoteToImmutable :: WriteQueue -> StrictTVar IO Bool -> [LeiosPoint] -> IO ()
sqlPromoteToImmutable :: WriteQueue -> StrictTVar IO Bool -> [LeiosPoint] -> IO ()
sqlPromoteToImmutable WriteQueue
writeQueue StrictTVar IO Bool
copierDoorbell [LeiosPoint]
points = Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([LeiosPoint] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [LeiosPoint]
points) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  Promise IO () -> IO ()
forall (m :: * -> *) a. Promise m a -> m a
await (Promise IO () -> IO ()) -> IO (Promise IO ()) -> IO ()
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< WriteQueue -> (WriteResult () -> WriteJob) -> IO (Promise IO ())
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue ([Hash HASH EraIndependentEbReferences]
-> WriteResult () -> WriteJob
PinEb [LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
p | LeiosPoint
p <- [LeiosPoint]
points])
  -- notify the copier thread that there's work to be done
  STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ StrictTVar IO Bool -> Bool -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO Bool
copierDoorbell Bool
True

-- | The copier's connection and its prepared statements.
data CopierConn = CopierConn
  { CopierConn -> Database
ccDb :: !DB.Database
  -- ^ main = immutable partition, @vol@ = attached volatile partition
  , CopierConn -> CopierStmts
ccStmts :: !CopierStmts
  -- ^ Precompiled statements used by the copier.
  }

-- | Copy one pinned EB's closure into the immutable partition. 'True' when
-- it landed, so the caller may have its volatile rows marked as copied.
copyEbToImmutable ::
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  CopierConn ->
  EndorserBlockHash ->
  IO Bool
copyEbToImmutable :: Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> CopierConn
-> Hash HASH EraIndependentEbReferences
-> IO Bool
copyEbToImmutable Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar CopierConn
conn Hash HASH EraIndependentEbReferences
ebHash =
  IO (Maybe Int)
appendToImmutable IO (Maybe Int) -> (Maybe Int -> IO Bool) -> IO Bool
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
    Maybe Int
Nothing -> do
      -- EB is not ready to be copied
      Tracer IO TraceLeiosDb -> TraceLeiosDb -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO TraceLeiosDb
tracer (TraceLeiosDb -> IO ()) -> TraceLeiosDb -> IO ()
forall a b. (a -> b) -> a -> b
$
        [Char] -> [Char] -> TraceLeiosDb
TraceLeiosDbCopyError
          (Hash HASH EraIndependentEbReferences -> [Char]
forall a. Show a => a -> [Char]
show Hash HASH EraIndependentEbReferences
ebHash)
          [Char]
"pinned EB has no complete closure in the volatile partition"
      Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
    Just Int
_copiedTxs -> do
      -- successfully copied, bump the stats
      StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpImmutableStats StrictTVar IO LeiosDbStats
statsVar Int
1
      Tracer IO TraceLeiosDb -> TraceLeiosDb -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO TraceLeiosDb
tracer (TraceLeiosDb -> IO ()) -> TraceLeiosDb -> IO ()
forall a b. (a -> b) -> a -> b
$ Int -> TraceLeiosDb
TraceLeiosDbCopiedToImmutable Int
1
      Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
 where
  CopierConn{Database
ccDb :: CopierConn -> Database
ccDb :: Database
ccDb, CopierStmts
ccStmts :: CopierConn -> CopierStmts
ccStmts :: CopierStmts
ccStmts} = CopierConn
conn
  CopierStmts{Statement
ccCompleteness :: Statement
ccCompleteness :: CopierStmts -> Statement
ccCompleteness, Statement
ccInsertEb :: Statement
ccInsertEb :: CopierStmts -> Statement
ccInsertEb, Statement
ccInsertEbTxs :: Statement
ccInsertEbTxs :: CopierStmts -> Statement
ccInsertEbTxs, Statement
ccInsertEbTxBytes :: Statement
ccInsertEbTxBytes :: CopierStmts -> Statement
ccInsertEbTxBytes} = CopierStmts
ccStmts

  -- Attempt to do the actual copying.
  --
  -- Returns the number of copied body rows, or Nothing if
  -- the volatile partition does not hold the full closure.
  appendToImmutable :: IO (Maybe Int)
  appendToImmutable :: IO (Maybe Int)
appendToImmutable =
    Database -> IO (Maybe Int) -> IO (Maybe Int)
forall a. HasCallStack => Database -> IO a -> IO a
dbWithTransaction Database
ccDb (IO (Maybe Int) -> IO (Maybe Int))
-> IO (Maybe Int) -> IO (Maybe Int)
forall a b. (a -> b) -> a -> b
$ do
      -- the cert-RB is on our chain, hence the EB must be complete by this point.
      -- Still check if it is, as a defensive programming measure, as it's very cheap.
      (bodyCount, closureCount) <-
        Statement -> IO (Int64, Int64) -> IO (Int64, Int64)
forall a. Statement -> IO a -> IO a
useStmt Statement
ccCompleteness (IO (Int64, Int64) -> IO (Int64, Int64))
-> IO (Int64, Int64) -> IO (Int64, Int64)
forall a b. (a -> b) -> a -> b
$ do
          Statement
-> ParamIndex -> Hash HASH EraIndependentEbReferences -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
ccCompleteness ParamIndex
1 Hash HASH EraIndependentEbReferences
ebHash
          -- step the first time, expecting a single result row
          HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStepSafe Statement
ccCompleteness IO StepResult
-> (StepResult -> IO (Int64, Int64)) -> IO (Int64, Int64)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            StepResult
DB.Done ->
              -- no row: critical error, fail fast and loud.
              -- this should not happen.
              [Char] -> IO (Int64, Int64)
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
[Char] -> m a
throwLeiosDbException [Char]
"sql_copy_completeness: expected a row"
            StepResult
DB.Row -> do
              n <- Statement -> ColumnIndex -> IO Int64
DB.columnInt64 Statement
ccCompleteness ColumnIndex
0
              m <- DB.columnInt64 ccCompleteness 1
              -- step again, expecting no more row
              dbStepSafe ccCompleteness >>= \case
                StepResult
DB.Done ->
                  -- we have our result
                  (Int64, Int64) -> IO (Int64, Int64)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int64
n, Int64
m)
                StepResult
DB.Row ->
                  -- another row: critical error, fail fast and loud.
                  -- this should not happen.
                  [Char] -> IO (Int64, Int64)
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
[Char] -> m a
throwLeiosDbException [Char]
"sql_copy_completeness: expected exactly one row"
      if bodyCount == 0 || bodyCount /= closureCount
        then
          -- the EB is incomplete, don't copy
          pure Nothing
        else do
          -- the EB is complete: body is non-empty and bodyCounty matches closureCount
          --
          -- copy the EB
          useStmt ccInsertEb $ do
            dbBindHash ccInsertEb 1 ebHash
            dbStep1Safe ccInsertEb
          -- copy the eb-to-transactions mapping
          useStmt ccInsertEbTxs $ do
            dbBindHash ccInsertEbTxs 1 ebHash
            dbStep1Safe ccInsertEbTxs
          nTxs <- DB.changes ccDb
          -- copy the tx bytes
          useStmt ccInsertEbTxBytes $ do
            dbBindHash ccInsertEbTxBytes 1 ebHash
            dbStep1Safe ccInsertEbTxBytes
          pure (Just nTxs)

-- | The copier: the only writer into the immutable partition.
--
-- Returns the copier's thread: cancelling it closes its connection.
startCopier ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  StrictTVar IO Bool ->
  WriteQueue ->
  FilePath ->
  FilePath ->
  IO (Thread IO ())
startCopier :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> StrictTVar IO Bool
-> WriteQueue
-> [Char]
-> [Char]
-> IO (Thread IO ())
startCopier ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar StrictTVar IO Bool
copierDoorbell WriteQueue
writeQueue [Char]
volPath [Char]
immPath =
  ResourceRegistry IO -> [Char] -> IO () -> IO (Thread IO ())
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, MonadEvaluate m,
 HasCallStack) =>
ResourceRegistry m -> [Char] -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry IO
registry [Char]
"leiosdb-copier" (IO () -> IO (Thread IO ())) -> IO () -> IO (Thread IO ())
forall a b. (a -> b) -> a -> b
$
    IO Database -> (Database -> IO ()) -> (Database -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (HasCallStack => [Char] -> IO Database
[Char] -> IO Database
openRawConnection [Char]
immPath) HasCallStack => Database -> IO ()
Database -> IO ()
closeChecked ((Database -> IO ()) -> IO ()) -> (Database -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Database
db -> do
      Database -> [Char] -> (Statement -> IO ()) -> IO ()
forall a.
HasCallStack =>
Database -> [Char] -> (Statement -> IO a) -> IO a
withStmt Database
db [Char]
"ATTACH ? AS vol" ((Statement -> IO ()) -> IO ()) -> (Statement -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Statement
stmt -> do
        HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindUtf8 Statement
stmt ParamIndex
1 ([Char] -> ByteString
forall a. IsString a => [Char] -> a
fromString [Char]
volPath)
        HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1Safe Statement
stmt
      IO CopierStmts
-> (CopierStmts -> IO ()) -> (CopierStmts -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (HasCallStack => Database -> IO CopierStmts
Database -> IO CopierStmts
prepareCopierStmts Database
db) CopierStmts -> IO ()
finalizeCopierStmts ((CopierStmts -> IO ()) -> IO ())
-> (CopierStmts -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \CopierStmts
stmts ->
        IO Statement
-> (Statement -> IO ()) -> (Statement -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (HasCallStack => Database -> Utf8 -> IO Statement
Database -> Utf8 -> IO Statement
dbPrepare Database
db ([Char] -> Utf8
forall a. IsString a => [Char] -> a
fromString [Char]
sql_next_pinned_eb)) Statement -> IO ()
dbFinalize ((Statement -> IO ()) -> IO ()) -> (Statement -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
          CopierConn -> Statement -> IO ()
runCopier (Database -> CopierStmts -> CopierConn
CopierConn Database
db CopierStmts
stmts)
 where
  runCopier ::
    CopierConn ->
    -- 'sql_next_pinned_eb'
    DB.Statement ->
    IO ()
  runCopier :: CopierConn -> Statement -> IO ()
runCopier CopierConn
copierConn Statement
nextPinnedStmt = IO ()
forall {b}. IO b
loop
   where
    CopierConn{Database
ccDb :: CopierConn -> Database
ccDb :: Database
ccDb} = CopierConn
copierConn

    nextPinnedBatch :: IO [Hash HASH a]
nextPinnedBatch = do
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
nextPinnedStmt ParamIndex
1 (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
copyBatchSize)
      Statement -> IO [Hash HASH a] -> IO [Hash HASH a]
forall a. Statement -> IO a -> IO a
useStmt Statement
nextPinnedStmt (IO [Hash HASH a] -> IO [Hash HASH a])
-> IO [Hash HASH a] -> IO [Hash HASH a]
forall a b. (a -> b) -> a -> b
$
        let rows :: [Hash h a] -> IO [Hash h a]
rows [Hash h a]
acc =
              HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStepSafe Statement
nextPinnedStmt IO StepResult -> (StepResult -> IO [Hash h a]) -> IO [Hash h a]
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
                StepResult
DB.Done -> [Hash h a] -> IO [Hash h a]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([Hash h a] -> [Hash h a]
forall a. [a] -> [a]
reverse [Hash h a]
acc)
                StepResult
DB.Row -> do
                  h <- Statement -> ColumnIndex -> IO (Hash h a)
forall h a.
(HasCallStack, HashAlgorithm h) =>
Statement -> ColumnIndex -> IO (Hash h a)
dbColumnHash Statement
nextPinnedStmt ColumnIndex
0
                  rows (h : acc)
         in [Hash HASH a] -> IO [Hash HASH a]
forall {h} {a}. HashAlgorithm h => [Hash h a] -> IO [Hash h a]
rows []

    -- 'True' when the EB is in the immutable partition and its pin can be
    -- retired. The EB stays pinned either way, so it is still next to copy.
    -- Backing off is what keeps a closure that never completes -- or a
    -- partition that keeps refusing the write -- from spinning here.
    copyOne :: Hash HASH EraIndependentEbReferences -> IO Bool
copyOne Hash HASH EraIndependentEbReferences
ebHash =
      Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> CopierConn
-> Hash HASH EraIndependentEbReferences
-> IO Bool
copyEbToImmutable Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar CopierConn
copierConn Hash HASH EraIndependentEbReferences
ebHash
        IO Bool -> (LeiosDbException -> IO Bool) -> IO Bool
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` \(LeiosDbException
e :: LeiosDbException) -> do
          _ <- Database -> Utf8 -> IO (Either (Error, Utf8) ())
DB.exec Database
ccDb Utf8
"ROLLBACK"
          traceWith tracer $ TraceLeiosDbCopyError (show ebHash) (displayException e)
          pure False

    -- One 'MarkCopied' for the batch: the mark is what makes a copied EB
    -- evictable, and under sync a queue round-trip per EB queues behind
    -- saturated inserts.
    copyBatch :: [Hash HASH EraIndependentEbReferences] -> IO ()
copyBatch [Hash HASH EraIndependentEbReferences]
ebHashes = do
      copied <- (Hash HASH EraIndependentEbReferences -> IO Bool)
-> [Hash HASH EraIndependentEbReferences]
-> IO [Hash HASH EraIndependentEbReferences]
forall (m :: * -> *) a.
Applicative m =>
(a -> m Bool) -> [a] -> m [a]
filterM Hash HASH EraIndependentEbReferences -> IO Bool
copyOne [Hash HASH EraIndependentEbReferences]
ebHashes
      if null copied
        then threadDelay copyRetryMicros
        else void . await =<< submitJob writeQueue (MarkCopied copied)

    loop :: IO b
loop = do
      -- Clear the copier doorbell before looking if there's any copying work to do,
      -- so a pin that lands while we look rings again instead of being lost.
      STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ StrictTVar IO Bool -> Bool -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO Bool
copierDoorbell Bool
False
      IO [Hash HASH EraIndependentEbReferences]
forall {a}. IO [Hash HASH a]
nextPinnedBatch IO [Hash HASH EraIndependentEbReferences]
-> ([Hash HASH EraIndependentEbReferences] -> IO b) -> IO b
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        batch :: [Hash HASH EraIndependentEbReferences]
batch@(Hash HASH EraIndependentEbReferences
_ : [Hash HASH EraIndependentEbReferences]
_) -> [Hash HASH EraIndependentEbReferences] -> IO ()
copyBatch [Hash HASH EraIndependentEbReferences]
batch IO () -> IO b -> IO b
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO b
loop
        [] -> do
          STM () -> IO ()
forall a. STM a -> IO a
IO.atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ StrictTVar IO Bool -> STM IO Bool
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO Bool
copierDoorbell STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
Bool -> STM IO ()
forall (m :: * -> *). MonadSTM m => Bool -> STM m ()
check
          IO b
loop

-- | How long the copier waits before trying a pinned EB again.
copyRetryMicros :: Int
copyRetryMicros :: Int
copyRetryMicros = Int
1000000

-- | How many pinned EBs the copier takes per pass, and so per 'MarkCopied'.
copyBatchSize :: Int
copyBatchSize :: Int
copyBatchSize = Int
32

-- * Garbage collection of the volatile partition

-- | How many EBs one sweep transaction takes, by default.
defaultGcBatchSize :: Int64
defaultGcBatchSize :: Int64
defaultGcBatchSize = Int64
4

-- | Implements 'Ouroboros.Consensus.Storage.LeiosDB.API.garbageCollect': the MARK phase of GC mark-and-sweep,
-- as a 'GcMark' job on the writer (see 'gcMark').
sqlGarbageCollect :: WriteQueue -> SlotNo -> IO ()
sqlGarbageCollect :: WriteQueue -> SlotNo -> IO ()
sqlGarbageCollect WriteQueue
writeQueue SlotNo
gcSlot =
  IO Int -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int -> IO ())
-> (Promise IO Int -> IO Int) -> Promise IO Int -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Promise IO Int -> IO Int
forall (m :: * -> *) a. Promise m a -> m a
await (Promise IO Int -> IO ()) -> IO (Promise IO Int) -> IO ()
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< WriteQueue -> (WriteResult Int -> WriteJob) -> IO (Promise IO Int)
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue (SlotNo -> WriteResult Int -> WriteJob
GcMark SlotNo
gcSlot)

-- | The MARK phase of GC mark-and-sweep: mark for GC (@status = 3@) every EB
-- hash all of whose announcements are older than the given slot and not
-- pinned (@status = 1@). Returns the number of 'ebs' rows marked.
--
-- Runs on the writer. It does not do much work, but rather primes the state
-- for the writer's own sweep steps (see 'startWriter'): the writer evicts the
-- marked EBs one batch per turn, so an insert waits for a batch at most.
gcMark ::
  HasCallStack =>
  DB.Database ->
  GcStmts ->
  SlotNo ->
  IO Int
gcMark :: HasCallStack => Database -> GcStmts -> SlotNo -> IO Int
gcMark Database
db GcStmts
gcStmts SlotNo
gcSlot = do
  let GcStmts{Statement
gsHasWork :: Statement
gsHasWork :: GcStmts -> Statement
gsHasWork, Statement
gsMarkEbForGC :: Statement
gsMarkEbForGC :: GcStmts -> Statement
gsMarkEbForGC} = GcStmts
gcStmts
  -- check if GC has any work to do
  hasWork <-
    Statement -> IO Bool -> IO Bool
forall a. Statement -> IO a -> IO a
useStmt Statement
gsHasWork (IO Bool -> IO Bool) -> IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
gsHasWork ParamIndex
1 Int64
slot
      (Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
/= Int64
0) (Int64 -> Bool) -> IO Int64 -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> HasCallStack => Statement -> IO Int64
Statement -> IO Int64
readSingleInt64 Statement
gsHasWork
  if hasWork
    then
      -- mark EBs that are ready to be garbage collected
      dbWithWriteTransactionRaw db $ do
        useStmt gsMarkEbForGC $ do
          dbBindInt64 gsMarkEbForGC 1 slot
          dbStep1Safe gsMarkEbForGC
        DB.changes db
    else pure 0
 where
  slot :: Int64
slot = Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (SlotNo -> Word64
unSlotNo SlotNo
gcSlot)

-- * Sweeping GC-marked rows out of the volatile partition

-- | The sweeper's connection and its prepared statements.
data SweeperConn = SweeperConn
  { SweeperConn -> Database
swDb :: !DB.Database
  -- ^ The writer's connection to the volatile partition.
  , SweeperConn -> SweeperStmts
swStmts :: !SweeperStmts
  -- ^ Precompiled statements used by the sweeper.
  }

-- | One 'SweepEbBatch' transaction: evict up to the given number of GC-marked
-- EBs, with their body and tx bytes -- three range deletes on the EBs' hashes.
-- Returns the number of evicted 'ebs' rows. Runs on the writer.
sweepEbBatch :: SweeperConn -> Int64 -> IO Int
sweepEbBatch :: SweeperConn -> Int64 -> IO Int
sweepEbBatch SweeperConn
conn Int64
batchSize = do
  let SweeperConn{Database
swDb :: SweeperConn -> Database
swDb :: Database
swDb, SweeperStmts
swStmts :: SweeperConn -> SweeperStmts
swStmts :: SweeperStmts
swStmts} = SweeperConn
conn
      SweeperStmts{Statement
swPickMarked :: Statement
swPickMarked :: SweeperStmts -> Statement
swPickMarked, Statement
swEvictEbTxBytes :: Statement
swEvictEbTxBytes :: SweeperStmts -> Statement
swEvictEbTxBytes, Statement
swEvictEbTxs :: Statement
swEvictEbTxs :: SweeperStmts -> Statement
swEvictEbTxs, Statement
swEvictEbs :: Statement
swEvictEbs :: SweeperStmts -> Statement
swEvictEbs} = SweeperStmts
swStmts
  Database -> IO Int -> IO Int
forall a. HasCallStack => Database -> IO a -> IO a
dbWithWriteTransactionRaw Database
swDb (IO Int -> IO Int) -> IO Int -> IO Int
forall a b. (a -> b) -> a -> b
$ do
    -- check if any EBs are ready to be evicted
    evictableEbs <- Statement -> IO [ByteString] -> IO [ByteString]
forall a. Statement -> IO a -> IO a
useStmt Statement
swPickMarked (IO [ByteString] -> IO [ByteString])
-> IO [ByteString] -> IO [ByteString]
forall a b. (a -> b) -> a -> b
$ do
      -- batchSize will never be negative, but we handle a negative batch size
      -- gracefully here with a negative LIMIT, which means no limit in SQLite.
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
swPickMarked ParamIndex
1 (if Int64
batchSize Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
<= Int64
0 then -Int64
1 else Int64
batchSize)
      HasCallStack => Statement -> IO [ByteString]
Statement -> IO [ByteString]
collectBlobs Statement
swPickMarked
    -- evict EBs if any are ready to be GCed
    if null evictableEbs
      then pure 0
      else do
        let evictableEbsJson = [ByteString] -> ByteString
jsonHexArray [ByteString]
evictableEbs
        execJson swEvictEbTxBytes evictableEbsJson
        execJson swEvictEbTxs evictableEbsJson
        -- last, so 'DB.changes' counts the evicted 'ebs' rows
        execJson swEvictEbs evictableEbsJson
        DB.changes swDb

-- * Stats

-- | Initialise 'LeiosDbStats' by counting the EB rows of both partitions.
--   This will only run once per process.
initialStats :: HasCallStack => FilePath -> FilePath -> IO LeiosDbStats
initialStats :: HasCallStack => [Char] -> [Char] -> IO LeiosDbStats
initialStats [Char]
volPath [Char]
immPath = do
  vol <- [Char] -> IO Int
forall {b}. Num b => [Char] -> IO b
countEbs [Char]
volPath
  imm <- countEbs immPath
  pure
    LeiosDbStats
      { volatileEbs = vol
      , immutableEbs = imm
      , walBytes = 0
      }
 where
  countEbs :: [Char] -> IO b
countEbs [Char]
path =
    Int64 -> b
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> b) -> IO Int64 -> IO b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Char] -> (Database -> IO Int64) -> IO Int64
forall a. HasCallStack => [Char] -> (Database -> IO a) -> IO a
withReadOnlyConn [Char]
path (\Database
db -> HasCallStack => Database -> [Char] -> IO Int64
Database -> [Char] -> IO Int64
queryInt64 Database
db [Char]
"SELECT COUNT(*) FROM ebs")

-- * Stats sampling

-- | Fork a thread that traces 'TraceLeiosDbStats' every 10 seconds. Samples
-- the volatile partition's file only.
startVolatileStatsSampler ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  FilePath ->
  IO (Thread IO ())
startVolatileStatsSampler :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> [Char]
-> IO (Thread IO ())
startVolatileStatsSampler ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar [Char]
volPath =
  ResourceRegistry IO -> [Char] -> IO () -> IO (Thread IO ())
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, MonadEvaluate m,
 HasCallStack) =>
ResourceRegistry m -> [Char] -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry IO
registry [Char]
"leiosdb-stats-sampler" (IO () -> IO (Thread IO ())) -> IO () -> IO (Thread IO ())
forall a b. (a -> b) -> a -> b
$ IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    -- wait one sample window to side-step contention with starting the LeiosDB
    Int -> IO ()
threadDelay Int
tenSeconds
    stats <- StrictTVar IO LeiosDbStats -> IO LeiosDbStats
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> m a
readTVarIO StrictTVar IO LeiosDbStats
statsVar
    -- read the WAL size
    walBytes <- fileSizeOr0 (volPath <> "-wal")
    traceWith tracer $
      TraceLeiosDbStats
        stats
          { walBytes
          }
 where
  tenSeconds :: Int
tenSeconds = Int
10000000

  fileSizeOr0 :: FilePath -> IO Integer
  fileSizeOr0 :: [Char] -> IO Integer
fileSizeOr0 [Char]
path = do
    exists <- [Char] -> IO Bool
doesFileExist [Char]
path
    if exists then getFileSize path else pure 0

-- | Fold a delta into the volatile EB count of the in-memory 'LeiosDbStats'.
bumpVolatileStats :: Conn -> Int -> IO ()
bumpVolatileStats :: Conn -> Int -> IO ()
bumpVolatileStats Conn{StrictTVar IO LeiosDbStats
connStats :: StrictTVar IO LeiosDbStats
connStats :: Conn -> StrictTVar IO LeiosDbStats
connStats} = StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpVolatileStatsVar StrictTVar IO LeiosDbStats
connStats

-- | 'bumpVolatileStats' for the maintenance paths, which have no 'Conn'.
bumpVolatileStatsVar :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpVolatileStatsVar :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpVolatileStatsVar StrictTVar IO LeiosDbStats
statsVar Int
dEbs =
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Int
dEbs Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      StrictTVar IO LeiosDbStats
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar IO LeiosDbStats
statsVar ((LeiosDbStats -> LeiosDbStats) -> STM IO ())
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall a b. (a -> b) -> a -> b
$
        \LeiosDbStats
s -> LeiosDbStats
s{volatileEbs = volatileEbs s + dEbs}

-- | Fold a delta into the immutable EB count of the in-memory 'LeiosDbStats'.
bumpImmutableStats :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpImmutableStats :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpImmutableStats StrictTVar IO LeiosDbStats
statsVar Int
dEbs =
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Int
dEbs Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      StrictTVar IO LeiosDbStats
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar IO LeiosDbStats
statsVar ((LeiosDbStats -> LeiosDbStats) -> STM IO ())
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall a b. (a -> b) -> a -> b
$
        \LeiosDbStats
s -> LeiosDbStats
s{immutableEbs = immutableEbs s + dEbs}