{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}

-- | The insert writes behind
-- 'Ouroboros.Consensus.Storage.LeiosDB.API.LeiosDbWriter'. They run on the
-- single writer, which calls them from its
-- 'Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.WriteQueue.WriteJob's.
module Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Insert
  ( sqlInsertEbPoint
  , sqlInsertEbBody
  , sqlInsertTxs
  ) where

import Cardano.Slotting.Slot (SlotNo (..))
import Control.Monad (forM_, when)
import Control.Tracer (Tracer)
import Data.ByteString (ByteString)
import qualified Data.ByteString as BS
import qualified Data.Set as Set
import qualified Database.SQLite3.Direct as DB
import Ouroboros.Consensus.Leios.Types
  ( BytesSize
  , EbHash (..)
  , LeiosEb
  , LeiosPoint (..)
  , TxHash (..)
  , encodeLeiosEbSize
  , leiosEbBodyItems
  )
import Ouroboros.Consensus.Storage.LeiosDB.API
  ( CompletedEbs
  , LeiosEbNotification (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.Exception (throwLeiosDbException)
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Connection
  ( Conn (..)
  , dbWithWriteTransaction
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Maintenance (bumpVolatileStats)
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Primitives
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Statements
import Ouroboros.Consensus.Storage.LeiosDB.Trace (TraceLeiosDb (..))

sqlInsertEbPoint :: Conn -> LeiosPoint -> BytesSize -> IO ()
sqlInsertEbPoint :: Conn -> LeiosPoint -> Word32 -> IO ()
sqlInsertEbPoint Conn
conn LeiosPoint
point Word32
ebBytesSize = do
  inserted <- Conn -> IO Int -> IO Int
forall a. HasCallStack => Conn -> IO a -> IO a
dbWithWriteTransaction Conn
conn (IO Int -> IO Int) -> IO Int -> IO Int
forall a b. (a -> b) -> a -> b
$ Statement -> IO Int -> IO Int
forall a. Statement -> IO a -> IO a
useStmt Statement
stmt (IO Int -> IO Int) -> IO Int -> IO Int
forall a b. (a -> b) -> a -> b
$ do
    HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stmt ParamIndex
1 (Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word64 -> Int64) -> Word64 -> Int64
forall a b. (a -> b) -> a -> b
$ SlotNo -> Word64
unSlotNo (LeiosPoint -> SlotNo
pointSlotNo LeiosPoint
point))
    HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stmt ParamIndex
2 (EbHash -> ByteString
ebHashBytes (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point))
    HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stmt ParamIndex
3 (Word32 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word32
ebBytesSize)
    HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1 Statement
stmt
    Database -> IO Int
DB.changes Database
db
  bumpVolatileStats conn inserted
 where
  Conn{conVolDb :: Conn -> Database
conVolDb = Database
db, connVolStmts :: Conn -> VolStmts
connVolStmts = VolStmts{stInsertEbPoint :: VolStmts -> Statement
stInsertEbPoint = Statement
stmt}} = Conn
conn

-- | Persist an EB body. The point MUST already be present (inserted
-- via 'sqlInsertEbPoint' on the announcement path).
sqlInsertEbBody ::
  Tracer IO TraceLeiosDb ->
  Conn ->
  (LeiosEbNotification -> IO ()) ->
  LeiosPoint ->
  LeiosEb ->
  IO CompletedEbs
sqlInsertEbBody :: Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> LeiosPoint
-> LeiosEb
-> IO CompletedEbs
sqlInsertEbBody Tracer IO TraceLeiosDb
tracer Conn
conn LeiosEbNotification -> IO ()
notify LeiosPoint
point LeiosEb
eb = do
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ([(Int, TxHash, Word32)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(Int, TxHash, Word32)]
items) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    [Char] -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
[Char] -> m a
throwLeiosDbException [Char]
"writeEbBody: empty EB body (programmer error)"
  completedNow <- Conn -> IO CompletedEbs -> IO CompletedEbs
forall a. HasCallStack => Conn -> IO a -> IO a
dbWithWriteTransaction Conn
conn (IO CompletedEbs -> IO CompletedEbs)
-> IO CompletedEbs -> IO CompletedEbs
forall a b. (a -> b) -> a -> b
$ do
    [(Int, TxHash, Word32)]
-> ((Int, TxHash, Word32) -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [(Int, TxHash, Word32)]
items (((Int, TxHash, Word32) -> IO ()) -> IO ())
-> ((Int, TxHash, Word32) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(Int
txOffset, TxHash
txHash, Word32
txBytesSize) -> Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
stInsertEbTxsRow (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInsertEbTxsRow ParamIndex
1 (EbHash -> ByteString
ebHashBytes (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point))
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInsertEbTxsRow ParamIndex
2 (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
txOffset)
      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInsertEbTxsRow ParamIndex
3 (let MkTxHash ByteString
bytes = TxHash
txHash in ByteString
bytes)
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInsertEbTxsRow ParamIndex
4 (Word32 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word32
txBytesSize)
      HasCallStack =>
Tracer IO TraceLeiosDb -> [Char] -> [Char] -> Statement -> IO ()
Tracer IO TraceLeiosDb -> [Char] -> [Char] -> Statement -> IO ()
dbStepInsertOrTrace
        Tracer IO TraceLeiosDb
tracer
        [Char]
"ebTxs"
        (EbHash -> [Char]
forall a. Show a => a -> [Char]
show (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
"@" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Int -> [Char]
forall a. Show a => a -> [Char]
show Int
txOffset)
        Statement
stInsertEbTxsRow
    -- Record which of this body's txs we still lack, then count them. Both in
    -- this transaction, so an arrival can never see the rows without the count
    -- or the other way round.
    Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
stInsertMissingTxs (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInsertMissingTxs ParamIndex
1 (EbHash -> ByteString
ebHashBytes (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point))
      HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1 Statement
stInsertMissingTxs
    -- Initialize missingTxCount and read the resulting value via
    -- @RETURNING missingTxCount@. Only /this/ point's row can have
    -- transitioned to 0 as a consequence of the insert above.
    missingCount <- Statement -> IO Int64 -> IO Int64
forall a. Statement -> IO a -> IO a
useStmt Statement
stInitMissingCount (IO Int64 -> IO Int64) -> IO Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ do
      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInitMissingCount ParamIndex
1 (EbHash -> ByteString
ebHashBytes (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point))
      HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInitMissingCount ParamIndex
2 (EbHash -> ByteString
ebHashBytes (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point))
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInitMissingCount ParamIndex
3 (Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word64 -> Int64) -> Word64 -> Int64
forall a b. (a -> b) -> a -> b
$ SlotNo -> Word64
unSlotNo (LeiosPoint -> SlotNo
pointSlotNo LeiosPoint
point))
      Statement -> IO Int64
readReturningInt64 Statement
stInitMissingCount
    if missingCount == 0
      then do
        useStmt stMarkPointNotified $ do
          dbBindInt64 stMarkPointNotified 1 (fromIntegral $ unSlotNo (pointSlotNo point))
          dbBindBlob stMarkPointNotified 2 (ebHashBytes (pointEbHash point))
          dbStep1 stMarkPointNotified
        pure [point]
      else pure []
  notify $ AcquiredEb point ebBytesSize
  forM_ completedNow $ \LeiosPoint
p -> LeiosEbNotification -> IO ()
notify (LeiosPoint -> LeiosEbNotification
AcquiredEbTxs LeiosPoint
p)
  pure completedNow
 where
  items :: [(Int, TxHash, Word32)]
items = LeiosEb -> [(Int, TxHash, Word32)]
leiosEbBodyItems LeiosEb
eb
  ebBytesSize :: Word32
ebBytesSize = LeiosEb -> Word32
encodeLeiosEbSize LeiosEb
eb
  Conn{VolStmts
connVolStmts :: Conn -> VolStmts
connVolStmts :: VolStmts
connVolStmts} = Conn
conn
  VolStmts
    { Statement
stInsertEbTxsRow :: Statement
stInsertEbTxsRow :: VolStmts -> Statement
stInsertEbTxsRow
    , Statement
stInsertMissingTxs :: Statement
stInsertMissingTxs :: VolStmts -> Statement
stInsertMissingTxs
    , Statement
stInitMissingCount :: Statement
stInitMissingCount :: VolStmts -> Statement
stInitMissingCount
    , Statement
stMarkPointNotified :: Statement
stMarkPointNotified :: VolStmts -> Statement
stMarkPointNotified
    } = VolStmts
connVolStmts

sqlInsertTxs ::
  Tracer IO TraceLeiosDb ->
  Conn ->
  (LeiosEbNotification -> IO ()) ->
  [(TxHash, ByteString)] ->
  IO CompletedEbs
sqlInsertTxs :: Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> [(TxHash, ByteString)]
-> IO CompletedEbs
sqlInsertTxs Tracer IO TraceLeiosDb
_tracer Conn
conn LeiosEbNotification -> IO ()
notify [(TxHash, ByteString)]
txs = do
  -- Skip txs already persisted in 'txs'. Under mempool backlog,
  -- successive forges (or overlapping peer EBs) re-present the same tx
  -- hashes; attempting the INSERT and catching a constraint violation
  -- still pays the bind + PK-lookup + reset cost per row.
  missing <- [TxHash] -> Set TxHash
forall a. Ord a => [a] -> Set a
Set.fromList ([TxHash] -> Set TxHash) -> IO [TxHash] -> IO (Set TxHash)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Conn -> [TxHash] -> IO [TxHash]
sqlFilterMissingTxs Conn
conn (((TxHash, ByteString) -> TxHash)
-> [(TxHash, ByteString)] -> [TxHash]
forall a b. (a -> b) -> [a] -> [b]
map (TxHash, ByteString) -> TxHash
forall a b. (a, b) -> a
fst [(TxHash, ByteString)]
txs)
  completed <- dbWithWriteTransaction conn $ do
    -- 'dbStepInsert' still handles the rare race where a concurrent
    -- writer inserted the same hash between the filter above and the
    -- INSERT below.
    forM_ (novel missing) $ \(TxHash
txHash, ByteString
txBytes) -> do
      let txBytesSize :: Int64
txBytesSize = Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Int64) -> Int -> Int64
forall a b. (a -> b) -> a -> b
$ ByteString -> Int
BS.length ByteString
txBytes
          txHashBytes :: ByteString
txHashBytes = let MkTxHash ByteString
bytes = TxHash
txHash in ByteString
bytes
      inserted <- Statement -> IO Bool -> IO Bool
forall a. Statement -> IO a -> IO a
useStmt Statement
stInsertTx (IO Bool -> IO Bool) -> IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
        HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInsertTx ParamIndex
1 ByteString
txHashBytes
        HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stInsertTx ParamIndex
2 ByteString
txBytes
        HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInsertTx ParamIndex
3 Int64
txBytesSize
        HasCallStack => Statement -> IO Bool
Statement -> IO Bool
dbStepInsert Statement
stInsertTx
      when inserted $ do
        useStmt stDecrMissingCount $ do
          dbBindBlob stDecrMissingCount 1 txHashBytes
          dbStep1 stDecrMissingCount
        -- Strictly after the decrement, which reads these rows.
        useStmt stDeleteMissingTxs $ do
          dbBindBlob stDeleteMissingTxs 1 txHashBytes
          dbStep1 stDeleteMissingTxs
    -- Find newly-complete EBs (missingTxCount reached 0)
    completed <- useStmt stFindCompleteEbs $ do
      let loop CompletedEbs
acc =
            HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStep Statement
stFindCompleteEbs IO StepResult -> (StepResult -> IO CompletedEbs) -> IO CompletedEbs
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 -> CompletedEbs -> IO CompletedEbs
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (CompletedEbs -> CompletedEbs
forall a. [a] -> [a]
reverse CompletedEbs
acc)
              StepResult
DB.Row -> do
                ebHash <- ByteString -> EbHash
MkEbHash (ByteString -> EbHash) -> IO ByteString -> IO EbHash
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Statement -> ColumnIndex -> IO ByteString
DB.columnBlob Statement
stFindCompleteEbs ColumnIndex
0
                slot <- SlotNo . fromIntegral <$> DB.columnInt64 stFindCompleteEbs 1
                loop (MkLeiosPoint slot ebHash : acc)
      loop []
    -- Mark them as notified so they are not found again
    useStmt stMarkNotifiedEbs $ dbStep1 stMarkNotifiedEbs
    pure completed
  -- Emit a closure-completion notification for each completed EB
  forM_ completed $ \LeiosPoint
point -> LeiosEbNotification -> IO ()
notify (LeiosPoint -> LeiosEbNotification
AcquiredEbTxs LeiosPoint
point)
  pure completed
 where
  Conn{VolStmts
connVolStmts :: Conn -> VolStmts
connVolStmts :: VolStmts
connVolStmts} = Conn
conn
  VolStmts
    { Statement
stInsertTx :: Statement
stInsertTx :: VolStmts -> Statement
stInsertTx
    , Statement
stDecrMissingCount :: Statement
stDecrMissingCount :: VolStmts -> Statement
stDecrMissingCount
    , Statement
stDeleteMissingTxs :: Statement
stDeleteMissingTxs :: VolStmts -> Statement
stDeleteMissingTxs
    , Statement
stFindCompleteEbs :: Statement
stFindCompleteEbs :: VolStmts -> Statement
stFindCompleteEbs
    , Statement
stMarkNotifiedEbs :: Statement
stMarkNotifiedEbs :: VolStmts -> Statement
stMarkNotifiedEbs
    } = VolStmts
connVolStmts
  novel :: Set TxHash -> [(TxHash, ByteString)]
novel Set TxHash
missing = ((TxHash, ByteString) -> Bool)
-> [(TxHash, ByteString)] -> [(TxHash, ByteString)]
forall a. (a -> Bool) -> [a] -> [a]
filter (\(TxHash
h, ByteString
_) -> TxHash
h TxHash -> Set TxHash -> Bool
forall a. Ord a => a -> Set a -> Bool
`Set.member` Set TxHash
missing) [(TxHash, ByteString)]
txs

-- | Batch-filter tx hashes against @txs@: passes txHashes as a JSON array
-- of hex strings; SQL decodes with @unhex()@ so index lookups on
-- @txs.txHashBytes@ still fire. Used internally by 'sqlInsertTxs' to skip
-- already-persisted txs.
sqlFilterMissingTxs :: Conn -> [TxHash] -> IO [TxHash]
sqlFilterMissingTxs :: Conn -> [TxHash] -> IO [TxHash]
sqlFilterMissingTxs Conn
conn [TxHash]
txHashes =
  Database -> IO [TxHash] -> IO [TxHash]
forall a. HasCallStack => Database -> IO a -> IO a
dbWithTransaction Database
db (IO [TxHash] -> IO [TxHash]) -> IO [TxHash] -> IO [TxHash]
forall a b. (a -> b) -> a -> b
$ Statement -> IO [TxHash] -> IO [TxHash]
forall a. Statement -> IO a -> IO a
useStmt Statement
stmt (IO [TxHash] -> IO [TxHash]) -> IO [TxHash] -> IO [TxHash]
forall a b. (a -> b) -> a -> b
$ do
    HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindUtf8 Statement
stmt ParamIndex
1 ([ByteString] -> ByteString
jsonHexArray [ByteString
b | MkTxHash ByteString
b <- [TxHash]
txHashes])
    [TxHash] -> IO [TxHash]
loop []
 where
  Conn{conVolDb :: Conn -> Database
conVolDb = Database
db, connVolStmts :: Conn -> VolStmts
connVolStmts = VolStmts{stFilterMissingTxs :: VolStmts -> Statement
stFilterMissingTxs = Statement
stmt}} = Conn
conn
  loop :: [TxHash] -> IO [TxHash]
loop [TxHash]
acc =
    HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStep Statement
stmt IO StepResult -> (StepResult -> IO [TxHash]) -> IO [TxHash]
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 -> [TxHash] -> IO [TxHash]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([TxHash] -> [TxHash]
forall a. [a] -> [a]
reverse [TxHash]
acc)
      StepResult
DB.Row -> do
        txHash <- ByteString -> TxHash
MkTxHash (ByteString -> TxHash) -> IO ByteString -> IO TxHash
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Statement -> ColumnIndex -> IO ByteString
DB.columnBlob Statement
stmt ColumnIndex
0
        loop (txHash : acc)