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

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

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

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

import Cardano.Prelude (forM_, when)
import Control.Concurrent.Class.MonadSTM.Strict
  ( StrictTChan
  , StrictTVar
  , check
  , dupTChan
  , isEmptyTBQueue
  , newBroadcastTChan
  , newTBQueueIO
  , newTVarIO
  , putTMVar
  , readTVar
  , readTVarIO
  , tryReadTBQueue
  , writeTChan
  , writeTVar
  )
import Control.Exception
  ( throwIO
  , toException
  )
import Control.Monad (unless, 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 qualified Database.SQLite3.Direct as DB
import qualified GHC.Conc as IO (atomically)
import qualified GHC.Stack
import Ouroboros.Consensus.Leios.Types (EbHash (..))
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.
--
-- Note that orphan transaction batch size is set by the 'gcOrphanTxBatchSize' constant.
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
  -- Both start set, so a restart picks up whatever the last run left
  -- pinned or marked; the copier and the writer clear them once they find
  -- nothing.
  copierDoorbell <- newTVarIO True
  sweepDoorbell <- newTVarIO True

  -- the volatile partition writer thread
  writeQueue <-
    startWriter
      registry
      tracer
      statsVar
      notificationChan
      sweepDoorbell
      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)
      , leiosDbGarbageCollect = 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 already reached the awaiter of the write
    -- that failed.
    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 -> LeiosEb -> IO (Promise IO [LeiosPoint])
writeEbBody = \LeiosPoint
point LeiosEb
eb -> WriteQueue
-> (WriteResult [LeiosPoint] -> WriteJob)
-> IO (Promise IO [LeiosPoint])
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue (LeiosPoint -> LeiosEb -> WriteResult [LeiosPoint] -> WriteJob
WriteEbBody LeiosPoint
point LeiosEb
eb)
        , writeTxs :: HasCallStack =>
[(TxHash, ByteString)] -> IO (Promise IO [LeiosPoint])
writeTxs = \[(TxHash, ByteString)]
txs -> WriteQueue
-> (WriteResult [LeiosPoint] -> WriteJob)
-> IO (Promise IO [LeiosPoint])
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue ([(TxHash, ByteString)] -> WriteResult [LeiosPoint] -> WriteJob
WriteTxs [(TxHash, 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 publishes each job's outcome into its 'WriteResult'. For insert
-- jobs it then rethrows a failure: publishing first lets an awaiting producer
-- see the exception rather than block on a promise the dying worker would
-- never fill, and a failed insert write is not survivable (no caller catches
-- 'LeiosDbException') -- producers still submitting eventually block on the
-- full queue and go down as blocked-indefinitely. A failed maintenance job is
-- the submitting scheduler's problem instead (traced, paced, retried); the
-- worker survives it.
startWriter ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  StrictTChan IO LeiosEbNotification ->
  StrictTVar IO Bool ->
  Int64 ->
  FilePath ->
  FilePath ->
  IO WriteQueue
startWriter :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> StrictTChan IO LeiosEbNotification
-> StrictTVar IO Bool
-> Int64
-> [Char]
-> [Char]
-> IO WriteQueue
startWriter ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar StrictTChan IO LeiosEbNotification
notificationChan StrictTVar IO Bool
sweepDoorbell 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
  sweepStateVar <- newTVarIO SweepIdle
  gcReinitDoneVar <- newTVarIO False
  jobsServedVar <- newTVarIO (0 :: Int)
  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
          } = IO ()
serve
         where
          runJob :: WriteJob -> IO Bool
          runJob :: WriteJob -> IO Bool
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)
              Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
            WriteEbPoint LeiosPoint
point BytesSize
size WriteResult ()
resultVar ->
              WriteResult () -> IO () -> IO ()
forall a. WriteResult a -> IO a -> IO ()
publish WriteResult ()
resultVar (Conn -> LeiosPoint -> BytesSize -> IO ()
sqlInsertEbPoint Conn
conn LeiosPoint
point BytesSize
size) IO () -> IO Bool -> IO Bool
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            WriteEbBody LeiosPoint
point LeiosEb
eb WriteResult [LeiosPoint]
resultVar ->
              WriteResult [LeiosPoint] -> IO [LeiosPoint] -> IO ()
forall a. WriteResult a -> IO a -> IO ()
publish WriteResult [LeiosPoint]
resultVar (Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> LeiosPoint
-> LeiosEb
-> IO [LeiosPoint]
sqlInsertEbBody Tracer IO TraceLeiosDb
tracer Conn
conn LeiosEbNotification -> IO ()
notify LeiosPoint
point LeiosEb
eb) IO () -> IO Bool -> IO Bool
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            WriteTxs [(TxHash, ByteString)]
txs WriteResult [LeiosPoint]
resultVar ->
              WriteResult [LeiosPoint] -> IO [LeiosPoint] -> IO ()
forall a. WriteResult a -> IO a -> IO ()
publish WriteResult [LeiosPoint]
resultVar (Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> [(TxHash, ByteString)]
-> IO [LeiosPoint]
sqlInsertTxs Tracer IO TraceLeiosDb
tracer Conn
conn LeiosEbNotification -> IO ()
notify [(TxHash, ByteString)]
txs) IO () -> IO Bool -> IO Bool
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            Flush WriteResult ()
resultVar ->
              WriteResult () -> IO () -> IO ()
forall a. WriteResult a -> IO a -> IO ()
publish WriteResult ()
resultVar (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()) IO () -> IO Bool -> IO Bool
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            PinEb [EbHash]
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.
              Database -> Database -> WriteResult () -> IO () -> IO ()
forall a. Database -> Database -> WriteResult a -> IO a -> IO ()
publishMaintenance Database
volDb Database
immDb 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
$
                  [EbHash] -> (EbHash -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [EbHash]
ebHashes ((EbHash -> IO ()) -> IO ()) -> (EbHash -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EbHash
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
                      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
pinStmt ParamIndex
1 (EbHash -> ByteString
ebHashBytes EbHash
ebHash)
                      HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1Safe Statement
pinStmt
              Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            MarkCopied [EbHash]
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.
              Database -> Database -> WriteResult () -> IO () -> IO ()
forall a. Database -> Database -> WriteResult a -> IO a -> IO ()
publishMaintenance Database
volDb Database
immDb 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
$
                  [EbHash] -> (EbHash -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [EbHash]
ebHashes ((EbHash -> IO ()) -> IO ()) -> (EbHash -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \EbHash
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
                      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
markCopiedStmt ParamIndex
1 (EbHash -> ByteString
ebHashBytes EbHash
ebHash)
                      HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1Safe Statement
markCopiedStmt
              Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            GcMark SlotNo
slot WriteResult ()
resultVar -> do
              Database -> Database -> WriteResult () -> IO () -> IO ()
forall a. Database -> Database -> WriteResult a -> IO a -> IO ()
publishMaintenance Database
volDb Database
immDb WriteResult ()
resultVar (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
                HasCallStack =>
StrictTVar IO Bool -> Database -> GcStmts -> SlotNo -> IO ()
StrictTVar IO Bool -> Database -> GcStmts -> SlotNo -> IO ()
gcMark StrictTVar IO Bool
sweepDoorbell Database
volDb GcStmts
gcStmts SlotNo
slot
              Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False

          -- Queued jobs first: writes arrive in bursts, and the quiet in
          -- between is what maintenance is for. But a burst can go on for as
          -- long as it likes, so 'maxJobsBetweenMaintenance' of them is the
          -- most that may pass before maintenance gets its turn regardless.
          serve :: IO ()
serve = do
            mJob <- 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
$ do
              served <- StrictTVar IO Int -> STM IO Int
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO Int
jobsServedVar
              if served >= maxJobsBetweenMaintenance
                then pure Nothing
                else
                  tryReadTBQueue queue >>= \case
                    Maybe WriteJob
Nothing -> Maybe WriteJob -> STM (Maybe WriteJob)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe WriteJob
forall a. Maybe a
Nothing
                    Just WriteJob
job -> WriteJob -> Maybe WriteJob
forall a. a -> Maybe a
Just WriteJob
job Maybe WriteJob -> STM () -> STM (Maybe WriteJob)
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ StrictTVar IO Int -> Int -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO Int
jobsServedVar (Int
served Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)
            case mJob of
              Just WriteJob
job -> do
                stop <- WriteJob -> IO Bool
runJob WriteJob
job
                traceWith tracer $ TraceLeiosDbWriteJobDone (describeJob job)
                unless stop serve
              Maybe WriteJob
Nothing -> do
                quiet <- IO Bool
stepMaintenance
                atomically $ writeTVar jobsServedVar 0
                when quiet blockUntilWork
                serve

          -- Nothing queued and no sweep outstanding: wait for either.
          blockUntilWork :: IO ()
blockUntilWork = 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
            rung <- readTVar sweepDoorbell
            check (not noJobs || rung)

          -- Advance a sweep by one batch; 'True' when there was nothing to do.
          -- 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.
          stepMaintenance :: IO Bool
stepMaintenance =
            IO Bool
step 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
volDb Utf8
"ROLLBACK"
              _ <- DB.exec immDb "ROLLBACK"
              traceWith tracer $ TraceLeiosDbGCError (displayException e)
              atomically $ writeTVar sweepStateVar SweepIdle
              pure True
           where
            step :: IO Bool
step =
              StrictTVar IO SweepState -> IO SweepState
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> m a
readTVarIO StrictTVar IO SweepState
sweepStateVar IO SweepState -> (SweepState -> 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
                SweepState
SweepIdle -> do
                  asked <- STM IO Bool -> IO Bool
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO Bool -> IO Bool) -> STM IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
                    rung <- StrictTVar IO Bool -> STM IO Bool
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO Bool
sweepDoorbell
                    when rung $ do
                      writeTVar sweepDoorbell False
                      writeTVar sweepStateVar (SweepEbs 0)
                    pure rung
                  if not asked
                    then pure True
                    else do
                      -- Stages the GC tx candidates a restart left behind.
                      done <- readTVarIO gcReinitDoneVar
                      unless done $ do
                        gcReinit sweeperConn
                        atomically $ writeTVar gcReinitDoneVar True
                      pure False
                SweepEbs Int
nEbs -> do
                  evicted <- SweeperConn -> Int64 -> IO Int
sweepEbBatch SweeperConn
sweeperConn Int64
gcBatchSize
                  if evicted == 0
                    then atomically $ writeTVar sweepStateVar (SweepOrphans nEbs 0)
                    else do
                      bumpVolatileStatsVar statsVar (negate evicted)
                      atomically $ writeTVar sweepStateVar (SweepEbs (nEbs + evicted))
                  pure False
                SweepOrphans Int
nEbs Int
nTxs ->
                  SweeperConn -> Int64 -> IO (Maybe Int)
sweepOrphanBatch SweeperConn
sweeperConn Int64
gcOrphanTxBatchSize 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
                    Just Int
evicted -> do
                      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 SweepState -> SweepState -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO SweepState
sweepStateVar (Int -> Int -> SweepState
SweepOrphans Int
nEbs (Int
nTxs Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
evicted))
                      Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
                    Maybe Int
Nothing -> do
                      Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
nEbs Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0 Bool -> Bool -> Bool
|| Int
nTxs Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                        -- Flush the WAL only after real work.
                        HasCallStack => Database -> Utf8 -> IO ()
Database -> Utf8 -> IO ()
dbExec Database
volDb Utf8
"PRAGMA wal_checkpoint(PASSIVE);"
                        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
TraceLeiosDbEvicted Int
nEbs
                      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 SweepState -> SweepState -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO SweepState
sweepStateVar SweepState
SweepIdle
                      Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False

      -- Seal, then fail what was already queued: nothing can be queued after
      -- the seal ('submitJob'), so afterwards the queue stays empty forever.
      sealAndDrain SomeException
cause = do
        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)
        let drain :: IO ()
drain =
              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 (StrictTBQueue IO WriteJob -> STM IO (Maybe WriteJob)
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> STM m (Maybe a)
tryReadTBQueue StrictTBQueue IO WriteJob
queue) IO (Maybe WriteJob) -> (Maybe WriteJob -> 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
                Maybe WriteJob
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                Just WriteJob
job -> SomeException -> WriteJob -> IO ()
failJob SomeException
cause WriteJob
job IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
drain
        IO ()
drain

  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, and the worker ends quietly.
      Just WriteResult ()
resultVar -> do
        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
$ WriteResult () -> Either SomeException () -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar WriteResult ()
resultVar Either SomeException ()
outcome
        SomeException -> IO ()
sealAndDrain SomeException
closedException
      -- 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 ()
sealAndDrain ((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
          }
  -- Only a 'LeiosDbException' is a failed write; anything else -- a
  -- cancellation above all -- belongs to this thread, not to the job.
  publish :: WriteResult a -> IO a -> IO ()
  publish :: forall a. WriteResult a -> IO a -> IO ()
publish WriteResult a
resultVar IO a
action =
    IO a -> IO (Either LeiosDbException a)
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 IO a
action IO (Either LeiosDbException a)
-> (Either LeiosDbException a -> 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
      Right a
x -> 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
$ WriteResult a -> Either SomeException a -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar WriteResult a
resultVar (a -> Either SomeException a
forall a b. b -> Either a b
Right a
x)
      Left (LeiosDbException
e :: LeiosDbException) -> do
        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
$ WriteResult a -> Either SomeException a -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar WriteResult a
resultVar (SomeException -> Either SomeException a
forall a b. a -> Either a b
Left (LeiosDbException -> SomeException
forall e. Exception e => e -> SomeException
toException LeiosDbException
e))
        LeiosDbException -> IO ()
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO LeiosDbException
e

  -- The transaction brackets have already rolled back on failure; the extra
  -- best-effort ROLLBACKs cover only the case where that rollback itself
  -- failed and would otherwise leave a transaction open under every later job.
  publishMaintenance :: DB.Database -> DB.Database -> WriteResult a -> IO a -> IO ()
  publishMaintenance :: forall a. Database -> Database -> WriteResult a -> IO a -> IO ()
publishMaintenance Database
volDb Database
immDb WriteResult a
resultVar IO a
action =
    IO a -> IO (Either LeiosDbException a)
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 IO a
action IO (Either LeiosDbException a)
-> (Either LeiosDbException a -> 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
      Right a
x -> 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
$ WriteResult a -> Either SomeException a -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar WriteResult a
resultVar (a -> Either SomeException a
forall a b. b -> Either a b
Right a
x)
      Left (LeiosDbException
e :: LeiosDbException) -> do
        _ <- Database -> Utf8 -> IO (Either (Error, Utf8) ())
DB.exec Database
volDb Utf8
"ROLLBACK"
        _ <- DB.exec immDb "ROLLBACK"
        atomically $ putTMVar resultVar (Left (toException e))