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

module Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite
  ( newLeiosDBSQLiteFromEnv
  , newLeiosDBSQLite
  , newLeiosDBSQLiteWithGcBatchSize
  , withLeiosDBSQLite

    -- * Re-exported for internal tooling
  , truncateLeiosDbAfterSlot
  , vacuumLeiosDb

    -- * SQL strings (re-exported for leios-schedule-gen)
  , sql_schema_vol
  , sql_schema_imm
  , sql_insert_eb
  , sql_insert_ebBody
  , sql_prealloc_ebTxBytes
  , sql_fill_ebTxBytes
  ) where

import Cardano.Prelude (forM_, when)
import Control.Concurrent.Class.MonadSTM.Strict
  ( StrictTChan
  , StrictTVar
  , check
  , dupTChan
  , isEmptyTBQueue
  , newBroadcastTChan
  , newTBQueueIO
  , newTVarIO
  , putTMVar
  , readTVarIO
  , tryReadTBQueue
  , writeTChan
  , writeTVar
  )
import Control.Exception
  ( throwIO
  , toException
  )
import Control.Monad (void)
import Control.Monad.Class.MonadThrow
  ( bracket
  , catch
  , displayException
  , try
  )
import Control.ResourceRegistry
  ( ResourceRegistry
  , Thread
  , cancelThread
  , forkLinkedThread
  , withRegistry
  )
import Control.Tracer (Tracer, traceWith)
import Data.IORef (newIORef, readIORef, writeIORef)
import Data.Int (Int64)
import Data.Maybe (fromMaybe)
import qualified Database.SQLite3.Direct as DB
import qualified GHC.Conc as IO (atomically)
import qualified GHC.Stack
import Ouroboros.Consensus.Storage.LeiosDB.API
  ( LeiosDbHandle (..)
  , LeiosDbReader (..)
  , LeiosDbWriter (..)
  , LeiosEbNotification (..)
  , Promise (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.Exception
  ( LeiosDbException (..)
  , LeiosDbFailure (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Connection
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Insert
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Maintenance
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Primitives
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Queries
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Read
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Schema
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Statements
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Tooling
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.WriteQueue
import Ouroboros.Consensus.Storage.LeiosDB.Trace (LeiosDbStats (..), TraceLeiosDb (..))
import Ouroboros.Consensus.Util.IOLike (atomically)
import System.Directory (createDirectoryIfMissing)
import System.Environment (lookupEnv)
import System.Exit (die)
import System.FilePath (takeDirectory)

-- * Public API

--- | Create a new Leios database connection from environment variable.
--- This looks up the LEIOS_VOL_DB_PATH and LEIOS_VOL_DB_PATH environment variables
--  and opens the database.
newLeiosDBSQLiteFromEnv :: ResourceRegistry IO -> Tracer IO TraceLeiosDb -> IO (LeiosDbHandle IO)
newLeiosDBSQLiteFromEnv :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb -> IO (LeiosDbHandle IO)
newLeiosDBSQLiteFromEnv ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer = do
  volDbPath <-
    [Char] -> IO (Maybe [Char])
lookupEnv [Char]
"LEIOS_VOL_DB_PATH" IO (Maybe [Char]) -> (Maybe [Char] -> IO [Char]) -> IO [Char]
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 [Char]
Nothing -> [Char] -> IO [Char]
forall a. [Char] -> IO a
die [Char]
"You must define the LEIOS_VOL_DB_PATH variable for this demo."
      Just [Char]
x -> [Char] -> IO [Char]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [Char]
x
  immDbPath <-
    lookupEnv "LEIOS_IMM_DB_PATH" >>= \case
      Maybe [Char]
Nothing -> [Char] -> IO [Char]
forall a. [Char] -> IO a
die [Char]
"You must define the LEIOS_IMM_DB_PATH variable for this demo."
      Just [Char]
x -> [Char] -> IO [Char]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [Char]
x
  newLeiosDBSQLite registry tracer volDbPath immDbPath

-- | Create a new Leios database using the SQLite implementation.
--
-- Each call to 'openReader' on the returned handle creates new SQLite
-- connection; readers are not thread-safe and should not be shared across
-- threads. All writers submit to the one write connection created here, on
-- its own worker thread.
--
-- The registry owns the background threads (writer, copier and a thread that
-- samples the database's size): 'closeLeiosDbHandle' stops them in order, and
-- closing the registry cancels whatever is still running.
--
-- Creates both files with their schemas before returning, so call it only
-- once their directories may be non-empty (after the ChainDB marker check).
newLeiosDBSQLite ::
  ResourceRegistry IO -> Tracer IO TraceLeiosDb -> FilePath -> FilePath -> IO (LeiosDbHandle IO)
newLeiosDBSQLite :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> [Char]
-> [Char]
-> IO (LeiosDbHandle IO)
newLeiosDBSQLite ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer [Char]
volLeiosDbPath [Char]
immLeiosDbPath =
  ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> [Char]
-> [Char]
-> Int64
-> IO (LeiosDbHandle IO)
newLeiosDBSQLiteWithGcBatchSize ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer [Char]
volLeiosDbPath [Char]
immLeiosDbPath Int64
defaultGcBatchSize

-- | 'newLeiosDBSQLite' with an explicit GC sweep batch size: how many EBs
-- the writer evicts per turn, between the jobs it serves.
newLeiosDBSQLiteWithGcBatchSize ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  FilePath ->
  FilePath ->
  Int64 ->
  IO (LeiosDbHandle IO)
newLeiosDBSQLiteWithGcBatchSize :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> [Char]
-> [Char]
-> Int64
-> IO (LeiosDbHandle IO)
newLeiosDBSQLiteWithGcBatchSize ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer [Char]
volLeiosDbPath [Char]
immLeiosDbPath Int64
gcBatchSize = do
  -- The database opens before whoever owns these directories creates them.
  ([Char] -> IO ()) -> [[Char]] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (Bool -> [Char] -> IO ()
createDirectoryIfMissing Bool
True ([Char] -> IO ()) -> ([Char] -> [Char]) -> [Char] -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Char] -> [Char]
takeDirectory) [[Char]
volLeiosDbPath, [Char]
immLeiosDbPath]
  -- create .db files
  HasCallStack => [Char] -> [Char] -> IO ()
[Char] -> [Char] -> IO ()
initialiseLeiosDbFiles [Char]
volLeiosDbPath [Char]
immLeiosDbPath
  notificationChan <- STM IO (StrictTChan IO LeiosEbNotification)
-> IO (StrictTChan IO LeiosEbNotification)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM IO (StrictTChan IO LeiosEbNotification)
forall (m :: * -> *) a. MonadSTM m => STM m (StrictTChan m a)
newBroadcastTChan
  -- seed the in-memory stats by counting the EB rows once per handle
  statsVar <- newTVarIO =<< initialStats volLeiosDbPath immLeiosDbPath
  -- start a thread to sample the sizes of the LeiosDB
  samplerThread <- startVolatileStatsSampler registry tracer statsVar volLeiosDbPath
  -- This variable signals the copier thread to wake up and copy the rows
  -- from the volatile to the immutable LeiosDB partition.
  -- It starts set, so that on a node restart we continue a potentially
  -- interrupted copy pass.
  copierDoorbell <- newTVarIO True

  -- the volatile partition writer thread
  writeQueue <-
    startWriter
      registry
      tracer
      statsVar
      notificationChan
      gcBatchSize
      volLeiosDbPath
      immLeiosDbPath
  -- the immutable partition writer thread
  copierThread <-
    startCopier
      registry
      tracer
      statsVar
      copierDoorbell
      writeQueue
      volLeiosDbPath
      immLeiosDbPath
  pure
    LeiosDbHandle
      { closeLeiosDbHandle = close samplerThread copierThread writeQueue
      , openReader = implOpenReader statsVar
      , openWriter = implOpenWriter writeQueue
      , subscribeEbNotifications = atomically (dupTChan notificationChan)
      , garbageCollect = sqlGarbageCollect writeQueue
      , leiosDbPromoteToImmutable = sqlPromoteToImmutable writeQueue copierDoorbell
      , leiosDbSampleStats = readTVarIO statsVar
      }
 where
  close :: Thread IO () -> Thread IO () -> WriteQueue -> IO ()
  close :: Thread IO () -> Thread IO () -> WriteQueue -> IO ()
close Thread IO ()
samplerThread Thread IO ()
copierThread WriteQueue
writeQueue = do
    -- The sampler holds no connection -- it only reads counters -- so
    -- cancelling it is safe at any point.
    Thread IO () -> IO ()
forall (m :: * -> *) a. MonadAsync m => Thread m a -> m ()
cancelThread Thread IO ()
samplerThread
    -- The copier before the writer: it submits to the writer, so it must be
    -- gone before the writer stops serving. Cancelling it rolls back any
    -- copy in flight; the EB stays pinned, so the next start copies it again.
    Thread IO () -> IO ()
forall (m :: * -> *) a. MonadAsync m => Thread m a -> m ()
cancelThread Thread IO ()
copierThread
    -- The queue is FIFO, so serving this job flushes everything
    -- submitted before it; awaiting it waits for the connections to
    -- close, and a failed close propagates -- a leaked connection must
    -- be loud.
    --
    -- A sealed queue refuses the job. Nothing is left to close then: the
    -- worker closes the connections on every exit path before it seals,
    -- and whatever killed it is already taking the node down through the
    -- link.
    IO (Promise IO ()) -> IO (Either LeiosDbException (Promise IO ()))
forall e a. Exception e => IO a -> IO (Either e a)
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> m (Either e a)
try (WriteQueue -> (WriteResult () -> WriteJob) -> IO (Promise IO ())
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue WriteResult () -> WriteJob
Shutdown) IO (Either LeiosDbException (Promise IO ()))
-> (Either LeiosDbException (Promise IO ()) -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Left (LeiosDbException
_writerGone :: LeiosDbException) -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      Right Promise IO ()
promise -> Promise IO () -> IO ()
forall (m :: * -> *) a. Promise m a -> m a
await Promise IO ()
promise

  implOpenReader :: StrictTVar IO LeiosDbStats -> IO (LeiosDbReader IO)
implOpenReader StrictTVar IO LeiosDbStats
statsVar = do
    volDb <- HasCallStack => [Char] -> IO Database
[Char] -> IO Database
openRawConnection [Char]
volLeiosDbPath
    immDb <- orCloseOnError volDb $ openRawConnection immLeiosDbPath
    conn <- mkConn tracer statsVar volDb immDb
    pure
      LeiosDbReader
        { closeReader = closeConn conn
        , testScanEbPoints = sqlScanEbPoints conn
        , scanCompleteEbClosuresNotOlderThanSlot = sqlScanCompleteEbPointsSince conn
        , lookupEbBody = sqlLookupEbBody conn
        , batchRetrieveTxs = sqlBatchRetrieveTxs conn
        , lookupEbClosure = sqlLookupEbClosure conn
        }

  implOpenWriter :: WriteQueue -> f (LeiosDbWriter IO)
implOpenWriter WriteQueue
writeQueue =
    LeiosDbWriter IO -> f (LeiosDbWriter IO)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
      LeiosDbWriter
        { -- Not a teardown -- the write connection outlives every writer.
          closeWriter :: IO ()
closeWriter = IO () -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO () -> IO ())
-> (Promise IO () -> IO ()) -> Promise IO () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. 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 WriteResult () -> WriteJob
Flush
        , writeEbPoint :: HasCallStack => LeiosPoint -> BytesSize -> IO (Promise IO ())
writeEbPoint = \LeiosPoint
point BytesSize
size -> WriteQueue -> (WriteResult () -> WriteJob) -> IO (Promise IO ())
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue (LeiosPoint -> BytesSize -> WriteResult () -> WriteJob
WriteEbPoint LeiosPoint
point BytesSize
size)
        , writeEbBody :: HasCallStack =>
LeiosPoint -> EndorserBlock -> IO (Promise IO [LeiosPoint])
writeEbBody = \LeiosPoint
point EndorserBlock
eb -> WriteQueue
-> (WriteResult [LeiosPoint] -> WriteJob)
-> IO (Promise IO [LeiosPoint])
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue (LeiosPoint -> EndorserBlock -> WriteResult [LeiosPoint] -> WriteJob
WriteEbBody LeiosPoint
point EndorserBlock
eb)
        , writeTxs :: HasCallStack =>
LeiosPoint -> [(Int, ByteString)] -> IO (Promise IO [LeiosPoint])
writeTxs = \LeiosPoint
point [(Int, ByteString)]
txs -> WriteQueue
-> (WriteResult [LeiosPoint] -> WriteJob)
-> IO (Promise IO [LeiosPoint])
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue (LeiosPoint
-> [(Int, ByteString)] -> WriteResult [LeiosPoint] -> WriteJob
WriteTxs LeiosPoint
point [(Int, ByteString)]
txs)
        }

-- | 'newLeiosDBSQLite' bracketed with its 'closeLeiosDbHandle': on release
-- every pending write has landed, the background threads are gone and the
-- connections are closed, so e.g. the database files can be deleted.
withLeiosDBSQLite ::
  Tracer IO TraceLeiosDb -> FilePath -> FilePath -> (LeiosDbHandle IO -> IO a) -> IO a
withLeiosDBSQLite :: forall a.
Tracer IO TraceLeiosDb
-> [Char] -> [Char] -> (LeiosDbHandle IO -> IO a) -> IO a
withLeiosDBSQLite Tracer IO TraceLeiosDb
tracer [Char]
volLeiosDbPath [Char]
immLeiosDbPath LeiosDbHandle IO -> IO a
k =
  (ResourceRegistry IO -> IO a) -> IO a
forall (m :: * -> *) a.
(MonadSTM m, MonadMask m, MonadThread m, MonadEvaluate m,
 HasCallStack) =>
(ResourceRegistry m -> m a) -> m a
withRegistry ((ResourceRegistry IO -> IO a) -> IO a)
-> (ResourceRegistry IO -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \ResourceRegistry IO
registry ->
    IO (LeiosDbHandle IO)
-> (LeiosDbHandle IO -> IO ())
-> (LeiosDbHandle IO -> IO a)
-> IO a
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
      (ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> [Char]
-> [Char]
-> IO (LeiosDbHandle IO)
newLeiosDBSQLite ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer [Char]
volLeiosDbPath [Char]
immLeiosDbPath)
      LeiosDbHandle IO -> IO ()
forall (m :: * -> *). LeiosDbHandle m -> m ()
closeLeiosDbHandle
      LeiosDbHandle IO -> IO a
k

-- | What the writer's worker owns while it serves: its connections to both
-- partitions and the statements prepared on them.
data WriterConns = WriterConns
  { WriterConns -> Database
wcVolDb :: !DB.Database
  , WriterConns -> Database
wcImmDb :: !DB.Database
  , WriterConns -> Conn
wcConn :: !Conn
  , WriterConns -> SweeperConn
wcSweeperConn :: !SweeperConn
  , WriterConns -> GcStmts
wcGcStmts :: !GcStmts
  , WriterConns -> Statement
wcPinStmt :: !DB.Statement
  -- ^ 'sql_pin_eb'
  , WriterConns -> Statement
wcMarkCopiedStmt :: !DB.Statement
  -- ^ 'sql_mark_as_copied'
  }

-- | Open the writer's connections, prepare its statements, and run the
-- action with them. Each resource has its own bracket, so whatever was
-- acquired is released on every way out, statements before connections: an
-- open statement holds the close off.
withWriterConns ::
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  FilePath ->
  FilePath ->
  (WriterConns -> IO a) ->
  IO a
withWriterConns :: forall a.
Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> [Char]
-> [Char]
-> (WriterConns -> IO a)
-> IO a
withWriterConns Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar [Char]
volPath [Char]
immPath WriterConns -> IO a
k =
  IO Database -> (Database -> IO ()) -> (Database -> IO a) -> IO a
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]
volPath) HasCallStack => Database -> IO ()
Database -> IO ()
closeChecked ((Database -> IO a) -> IO a) -> (Database -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Database
volDb ->
    IO Database -> (Database -> IO ()) -> (Database -> IO a) -> IO a
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 a) -> IO a) -> (Database -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Database
immDb ->
      IO Conn -> (Conn -> IO ()) -> (Conn -> IO a) -> IO a
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 (Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats -> Database -> Database -> IO Conn
mkConn Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar Database
volDb Database
immDb) Conn -> IO ()
finalizeConnStmts ((Conn -> IO a) -> IO a) -> (Conn -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Conn
conn ->
        IO SweeperStmts
-> (SweeperStmts -> IO ()) -> (SweeperStmts -> IO a) -> IO a
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 SweeperStmts
Database -> IO SweeperStmts
prepareSweeperStmts Database
volDb) SweeperStmts -> IO ()
finalizeSweeperStmts ((SweeperStmts -> IO a) -> IO a) -> (SweeperStmts -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \SweeperStmts
sweeperStmts ->
          IO GcStmts -> (GcStmts -> IO ()) -> (GcStmts -> IO a) -> IO a
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 GcStmts
Database -> IO GcStmts
prepareGcStmts Database
volDb) GcStmts -> IO ()
finalizeGcStmts ((GcStmts -> IO a) -> IO a) -> (GcStmts -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \GcStmts
gcStmts ->
            Database -> [Char] -> (Statement -> IO a) -> IO a
forall a.
HasCallStack =>
Database -> [Char] -> (Statement -> IO a) -> IO a
withStmt Database
volDb [Char]
sql_pin_eb ((Statement -> IO a) -> IO a) -> (Statement -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Statement
pinStmt ->
              Database -> [Char] -> (Statement -> IO a) -> IO a
forall a.
HasCallStack =>
Database -> [Char] -> (Statement -> IO a) -> IO a
withStmt Database
volDb [Char]
sql_mark_as_copied ((Statement -> IO a) -> IO a) -> (Statement -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Statement
markCopiedStmt ->
                WriterConns -> IO a
k
                  WriterConns
                    { wcVolDb :: Database
wcVolDb = Database
volDb
                    , wcImmDb :: Database
wcImmDb = Database
immDb
                    , wcConn :: Conn
wcConn = Conn
conn
                    , wcSweeperConn :: SweeperConn
wcSweeperConn = Database -> SweeperStmts -> SweeperConn
SweeperConn Database
volDb SweeperStmts
sweeperStmts
                    , wcGcStmts :: GcStmts
wcGcStmts = GcStmts
gcStmts
                    , wcPinStmt :: Statement
wcPinStmt = Statement
pinStmt
                    , wcMarkCopiedStmt :: Statement
wcMarkCopiedStmt = Statement
markCopiedStmt
                    }

-- | Start the worker draining the write queue. The worker opens the write
-- connections itself, on its own thread ('withWriterConns'), and closes them
-- on every way out.
--
-- The worker puts each job's result into its 'WriteResult'. A failed write is
-- not survivable: the worker stops and seals the queue with the exception, so
-- awaiting a job it did not serve, or submitting another, throws it.
startWriter ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  StrictTChan IO LeiosEbNotification ->
  Int64 ->
  FilePath ->
  FilePath ->
  IO WriteQueue
startWriter :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> StrictTChan IO LeiosEbNotification
-> Int64
-> [Char]
-> [Char]
-> IO WriteQueue
startWriter ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar StrictTChan IO LeiosEbNotification
notificationChan Int64
gcBatchSize [Char]
volPath [Char]
immPath = do
  queue <- Natural -> IO (StrictTBQueue IO WriteJob)
forall (m :: * -> *) a.
MonadSTM m =>
Natural -> m (StrictTBQueue m a)
newTBQueueIO Natural
writerQueueDepth
  sealedVar <- newTVarIO Nothing
  -- Set by a served 'Shutdown', whose awaiter gets the outcome of the close.
  shutdownVar <- newIORef Nothing
  let notify = 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 ())
-> (LeiosEbNotification -> STM IO ())
-> LeiosEbNotification
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StrictTChan IO LeiosEbNotification
-> LeiosEbNotification -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTChan m a -> a -> STM m ()
writeTChan StrictTChan IO LeiosEbNotification
notificationChan

      worker :: WriterConns -> IO ()
      worker
        WriterConns
          { wcVolDb :: WriterConns -> Database
wcVolDb = Database
volDb
          , wcImmDb :: WriterConns -> Database
wcImmDb = Database
immDb
          , wcConn :: WriterConns -> Conn
wcConn = Conn
conn
          , wcSweeperConn :: WriterConns -> SweeperConn
wcSweeperConn = SweeperConn
sweeperConn
          , wcGcStmts :: WriterConns -> GcStmts
wcGcStmts = GcStmts
gcStmts
          , wcPinStmt :: WriterConns -> Statement
wcPinStmt = Statement
pinStmt
          , wcMarkCopiedStmt :: WriterConns -> Statement
wcMarkCopiedStmt = Statement
markCopiedStmt
          } = Int -> Maybe Int -> IO ()
serve Int
0 (Int -> Maybe Int
forall a. a -> Maybe a
Just Int
0)
         where
          -- The write worker takes care of two types of jobs: queued jobs (values of `WriteJob`)
          -- and maintenance jobs.
          -- We prioritise queued jobs first: writes arrive in bursts, and the quiet in
          -- between is the time to do maintenance. But a burst can go on for as
          -- long as it likes, so 'maxJobsBetweenMaintenance' prevents maintenance starvation.
          --
          -- The second argument is the state of the GC sweep pass: 'Nothing'
          -- when no sweep is in progress; 'Just' the EBs evicted so far while a
          -- pass is in progress. It starts as 'Just', to continue a sweep pass
          -- potentially interrupted by a node restart.
          serve :: Int -> Maybe Int -> IO ()
          serve :: Int -> Maybe Int -> IO ()
serve Int
served Maybe Int
sweep = do
            mJob <-
              if Int
served Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
maxJobsBetweenMaintenance
                then
                  -- we've served enough queued jobs, time to do maintenance
                  Maybe WriteJob -> IO (Maybe WriteJob)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe WriteJob
forall a. Maybe a
Nothing
                else
                  -- 'Nothing' when there are no queued jobs: do maintenance
                  STM IO (Maybe WriteJob) -> IO (Maybe WriteJob)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO (Maybe WriteJob) -> IO (Maybe WriteJob))
-> STM IO (Maybe WriteJob) -> IO (Maybe WriteJob)
forall a b. (a -> b) -> a -> b
$ StrictTBQueue IO WriteJob -> STM IO (Maybe WriteJob)
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> STM m (Maybe a)
tryReadTBQueue StrictTBQueue IO WriteJob
queue
            case mJob of
              -- do a normal queue job
              Just WriteJob
job -> do
                next <- WriteJob -> IO WhetherContinueServing
runJob WriteJob
job
                traceWith tracer $ TraceLeiosDbWriteJobDone (describeJob job)
                case next of
                  WhetherContinueServing
Stop -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                  WhetherContinueServing
Continue -> Int -> Maybe Int -> IO ()
serve (Int
served Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1) Maybe Int
sweep
                  WhetherContinueServing
ContinueAndSweep -> Int -> Maybe Int -> IO ()
serve (Int
served Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1) (Int -> Maybe Int
forall a. a -> Maybe a
Just (Int -> Maybe Int -> Int
forall a. a -> Maybe a -> a
fromMaybe Int
0 Maybe Int
sweep))
              -- do maintenance
              Maybe WriteJob
Nothing -> do
                sweep' <- Maybe Int -> IO (Maybe Int)
stepMaintenance Maybe Int
sweep
                case sweep' of
                  Maybe Int
Nothing -> IO ()
blockUntilJob
                  Just Int
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                serve 0 sweep'

          runJob :: WriteJob -> IO WhetherContinueServing
          runJob :: WriteJob -> IO WhetherContinueServing
runJob = \case
            Shutdown WriteResult ()
resultVar -> do
              -- Only stop serving: the brackets in 'withWriterConns' close the
              -- connections once 'serve' returns, and the worker hands the
              -- outcome of that close to this job's awaiter.
              IORef (Maybe (WriteResult ())) -> Maybe (WriteResult ()) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Maybe (WriteResult ()))
shutdownVar (WriteResult () -> Maybe (WriteResult ())
forall a. a -> Maybe a
Just WriteResult ()
resultVar)
              WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Stop
            WriteEbPoint LeiosPoint
point BytesSize
size WriteResult ()
resultVar ->
              WriteResult () -> IO () -> IO ()
forall a. WriteResult a -> IO a -> IO a
publish WriteResult ()
resultVar (Conn -> LeiosPoint -> BytesSize -> IO ()
sqlInsertEbPoint Conn
conn LeiosPoint
point BytesSize
size) IO () -> IO WhetherContinueServing -> IO WhetherContinueServing
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Continue
            WriteEbBody LeiosPoint
point EndorserBlock
eb WriteResult [LeiosPoint]
resultVar ->
              WriteResult [LeiosPoint] -> IO [LeiosPoint] -> IO [LeiosPoint]
forall a. WriteResult a -> IO a -> IO a
publish WriteResult [LeiosPoint]
resultVar (Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> LeiosPoint
-> EndorserBlock
-> IO [LeiosPoint]
sqlInsertEbBody Tracer IO TraceLeiosDb
tracer Conn
conn LeiosEbNotification -> IO ()
notify LeiosPoint
point EndorserBlock
eb) IO [LeiosPoint]
-> IO WhetherContinueServing -> IO WhetherContinueServing
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Continue
            WriteTxs LeiosPoint
point [(Int, ByteString)]
txs WriteResult [LeiosPoint]
resultVar ->
              WriteResult [LeiosPoint] -> IO [LeiosPoint] -> IO [LeiosPoint]
forall a. WriteResult a -> IO a -> IO a
publish WriteResult [LeiosPoint]
resultVar (Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> LeiosPoint
-> [(Int, ByteString)]
-> IO [LeiosPoint]
sqlInsertTxs Tracer IO TraceLeiosDb
tracer Conn
conn LeiosEbNotification -> IO ()
notify LeiosPoint
point [(Int, ByteString)]
txs) IO [LeiosPoint]
-> IO WhetherContinueServing -> IO WhetherContinueServing
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Continue
            Flush WriteResult ()
resultVar ->
              WriteResult () -> IO () -> IO ()
forall a. WriteResult a -> IO a -> IO a
publish WriteResult ()
resultVar (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()) IO () -> IO WhetherContinueServing -> IO WhetherContinueServing
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Continue
            PinEb [EndorserBlockHash]
ebHashes WriteResult ()
resultVar -> do
              -- One transaction for the batch: the submitter awaits this while
              -- holding the ImmutableDB write lock, so a round-trip per EB throttles
              -- block immutalisation to this queue's drain rate.
              WriteResult () -> IO () -> IO ()
forall a. WriteResult a -> IO a -> IO a
publish WriteResult ()
resultVar (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                Database -> IO () -> IO ()
forall a. HasCallStack => Database -> IO a -> IO a
dbWithWriteTransactionRaw Database
volDb (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                  [EndorserBlockHash] -> (EndorserBlockHash -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [EndorserBlockHash]
ebHashes ((EndorserBlockHash -> IO ()) -> IO ())
-> (EndorserBlockHash -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EndorserBlockHash
ebHash ->
                    Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
pinStmt (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                      Statement -> ParamIndex -> EndorserBlockHash -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
pinStmt ParamIndex
1 EndorserBlockHash
ebHash
                      HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1Safe Statement
pinStmt
              WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Continue
            MarkCopied [EndorserBlockHash]
ebHashes WriteResult ()
resultVar -> do
              -- One transaction for the batch: the mark is what makes a copied EB
              -- evictable, and under sync it otherwise costs a queue round-trip
              -- per EB behind a saturated insert FIFO.
              WriteResult () -> IO () -> IO ()
forall a. WriteResult a -> IO a -> IO a
publish WriteResult ()
resultVar (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                Database -> IO () -> IO ()
forall a. HasCallStack => Database -> IO a -> IO a
dbWithWriteTransactionRaw Database
volDb (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                  [EndorserBlockHash] -> (EndorserBlockHash -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [EndorserBlockHash]
ebHashes ((EndorserBlockHash -> IO ()) -> IO ())
-> (EndorserBlockHash -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EndorserBlockHash
ebHash ->
                    Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
markCopiedStmt (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                      Statement -> ParamIndex -> EndorserBlockHash -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
markCopiedStmt ParamIndex
1 EndorserBlockHash
ebHash
                      HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1Safe Statement
markCopiedStmt
              WhetherContinueServing -> IO WhetherContinueServing
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WhetherContinueServing
Continue
            GcMark SlotNo
slot WriteResult Int
resultVar -> do
              nEbsMarked <- WriteResult Int -> IO Int -> IO Int
forall a. WriteResult a -> IO a -> IO a
publish WriteResult Int
resultVar (HasCallStack => Database -> GcStmts -> SlotNo -> IO Int
Database -> GcStmts -> SlotNo -> IO Int
gcMark Database
volDb GcStmts
gcStmts SlotNo
slot)
              pure $ if nEbsMarked > 0 then ContinueAndSweep else Continue

          -- Nothing queued and no sweep outstanding: wait for a job.
          blockUntilJob :: IO ()
blockUntilJob = STM () -> IO ()
forall a. STM a -> IO a
IO.atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
            noJobs <- StrictTBQueue IO WriteJob -> STM IO Bool
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> STM m Bool
isEmptyTBQueue StrictTBQueue IO WriteJob
queue
            check (not noJobs)

          -- Advance a sweep by one batch, returning the new sweep state.
          -- A failure here is the maintenance's problem, not the writer's:
          -- trace it, roll back anything left open, and let the next GC tick
          -- bring the work back. Hence 'catch' rather than 'bracket': the
          -- exception must stop here, not take the writer down, and the
          -- cleanup is for failure only.
          stepMaintenance :: Maybe Int -> IO (Maybe Int)
          stepMaintenance :: Maybe Int -> IO (Maybe Int)
stepMaintenance Maybe Int
sweep =
            IO (Maybe Int)
step IO (Maybe Int)
-> (LeiosDbException -> IO (Maybe Int)) -> IO (Maybe Int)
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
volDb Utf8
"ROLLBACK"
              _ <- DB.exec immDb "ROLLBACK"
              traceWith tracer $ TraceLeiosDbGCError (displayException e)
              pure Nothing
           where
            step :: IO (Maybe Int)
step =
              case Maybe Int
sweep of
                Maybe Int
Nothing -> Maybe Int -> IO (Maybe Int)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe Int
forall a. Maybe a
Nothing
                Just Int
nEbs -> do
                  evicted <- SweeperConn -> Int64 -> IO Int
sweepEbBatch SweeperConn
sweeperConn Int64
gcBatchSize
                  if evicted == 0
                    then do
                      when (nEbs > 0) $ do
                        -- Flush the WAL only after real work.
                        dbExec volDb "PRAGMA wal_checkpoint(PASSIVE);"
                        traceWith tracer $ TraceLeiosDbEvicted nEbs
                      pure Nothing
                    else do
                      bumpVolatileStatsVar statsVar (negate evicted)
                      pure (Just (nEbs + evicted))

      -- Nothing can be queued after the seal ('submitJob'), and awaiting a
      -- job left in the queue throws the cause.
      seal SomeException
cause = 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 (Maybe SomeException)
-> Maybe SomeException -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO (Maybe SomeException)
sealedVar (SomeException -> Maybe SomeException
forall a. a -> Maybe a
Just SomeException
cause)

  void $ forkLinkedThread registry "leiosdb-writer" $ do
    -- The brackets close the connections on every way out of 'worker': a
    -- served 'Shutdown', a failed write, a cancellation.
    outcome <- try $ withWriterConns tracer statsVar volPath immPath worker
    readIORef shutdownVar >>= \case
      -- Stopped on request: the close outcome, failed or not, is the
      -- awaiter's to report (a failed close through the seal), and the worker
      -- ends quietly.
      Just WriteResult ()
resultVar -> do
        (SomeException -> IO ())
-> (() -> IO ()) -> Either SomeException () -> IO ()
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (IO () -> SomeException -> IO ()
forall a b. a -> b -> a
const (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())) (STM () -> IO ()
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 ()) -> (() -> STM ()) -> () -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WriteResult () -> () -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar WriteResult ()
resultVar) Either SomeException ()
outcome
        SomeException -> IO ()
seal ((SomeException -> SomeException)
-> (() -> SomeException)
-> Either SomeException ()
-> SomeException
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either SomeException -> SomeException
forall a. a -> a
id (SomeException -> () -> SomeException
forall a b. a -> b -> a
const SomeException
closedException) Either SomeException ()
outcome)
      -- Anything else stopped the worker. Then out, so the link takes the
      -- node down where the write failed rather than at whatever submits
      -- next. Cancellation by the registry is the one stop the link lets pass.
      Maybe (WriteResult ())
Nothing -> do
        SomeException -> IO ()
seal ((SomeException -> SomeException)
-> (() -> SomeException)
-> Either SomeException ()
-> SomeException
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either SomeException -> SomeException
forall a. a -> a
id (SomeException -> () -> SomeException
forall a b. a -> b -> a
const SomeException
closedException) Either SomeException ()
outcome)
        (SomeException -> IO ())
-> (() -> IO ()) -> Either SomeException () -> IO ()
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either SomeException -> IO ()
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Either SomeException ()
outcome
  pure WriteQueue{wqJobs = queue, wqSealed = sealedVar, wqTracer = tracer}
 where
  closedException :: SomeException
closedException =
    LeiosDbException -> SomeException
forall e. Exception e => e -> SomeException
toException (LeiosDbException -> SomeException)
-> LeiosDbException -> SomeException
forall a b. (a -> b) -> a -> b
$
      LeiosDbFailure -> LeiosDbException
LeiosDbException
        LeiosDbFailure
          { ldfErrorMessage :: [Char]
ldfErrorMessage = [Char]
"the LeiosDB writer is closed"
          , ldfCallStack :: [Char]
ldfCallStack = CallStack -> [Char]
GHC.Stack.prettyCallStack CallStack
HasCallStack => CallStack
GHC.Stack.callStack
          }
  -- The action rolls back its own transaction on failure.
  publish :: WriteResult a -> IO a -> IO a
  publish :: forall a. WriteResult a -> IO a -> IO a
publish WriteResult a
resultVar IO a
action = do
    x <- IO a
action
    atomically $ putTMVar resultVar x
    pure x

-- | What the writer does after serving a job.
data WhetherContinueServing
  = -- | Stop serving.
    Stop
  | -- | Keep serving.
    Continue
  | -- | Keep serving, and start a GC sweep pass or extend the one in progress.
    ContinueAndSweep