{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE ScopedTypeVariables #-}

module Ouroboros.Consensus.Storage.LeiosDB.Impl.InMemory
  ( InMemoryLeiosDb (..)
  , emptyInMemoryLeiosDb
  , newLeiosDBInMemory
  , newLeiosDBInMemoryWith
  ) where

import Cardano.Prelude (Generic, forM_, maybeToList, when)
import Cardano.Slotting.Slot (SlotNo (..))
import Control.Concurrent.Class.MonadSTM.Strict
  ( StrictTChan
  , StrictTVar
  , dupTChan
  , modifyTVar
  , newBroadcastTChan
  , newTVarIO
  , readTVar
  , writeTChan
  )
import Data.ByteString (ByteString)
import qualified Data.ByteString as BS
import Data.IntMap.Strict (IntMap)
import qualified Data.IntMap.Strict as IntMap
import Data.Map.Strict (Map)
import qualified Data.Map.Strict as Map
import Data.Set (Set)
import qualified Data.Set as Set
import GHC.Stack (HasCallStack)
import qualified GHC.Stack as GHC
import Ouroboros.Consensus.Leios.Types
  ( BytesSize
  , EbHash (..)
  , LeiosEb
  , LeiosPoint (..)
  , TxHash (..)
  , encodeLeiosEbSize
  , leiosEbBodyItems
  )
import Ouroboros.Consensus.Storage.LeiosDB.API
  ( CompletedEbs
  , LeiosDbHandle (..)
  , LeiosDbReader (..)
  , LeiosDbStats (..)
  , LeiosDbWriter (..)
  , LeiosEbNotification (..)
  , Promise (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.Exception
  ( LeiosDbException (LeiosDbWriteException)
  , LeiosDbWriteFailure (..)
  , throwLeiosDbException
  )
import Ouroboros.Consensus.Util.IOLike
  ( IOLike
  , NoThunks (..)
  , atomically
  , throwIO
  , toException
  , try
  )

-- | In-memory database state.
data InMemoryLeiosDb = InMemoryLeiosDb
  { InMemoryLeiosDb -> Map TxHash (ByteString, BytesSize)
imTxs :: !(Map TxHash (ByteString, BytesSize))
  -- ^ Global transaction storage.
  , InMemoryLeiosDb -> Map LeiosPoint BytesSize
imEbPoints :: !(Map LeiosPoint BytesSize)
  -- ^ Inserted EB points with their expected sizes. The same EB
  -- content can be inserted at multiple slots; each
  -- @(slot, hash)@ point has its own entry and completes
  -- independently.
  , InMemoryLeiosDb -> Set LeiosPoint
imEbBodiesDownloaded :: !(Set LeiosPoint)
  -- ^ Subset of 'imEbPoints' for which 'writeEbBody' has been
  -- called. Only these points are considered for completion.
  , InMemoryLeiosDb -> Map EbHash (IntMap EbTxEntry)
imEbBodies :: !(Map EbHash (IntMap {- txOffset -} EbTxEntry))
  -- ^ EB tx-hash list, keyed by content hash. Shared across every
  -- point that references the same EB.
  , InMemoryLeiosDb -> Set LeiosPoint
imCompletedEbs :: !(Set LeiosPoint)
  -- ^ Points for which 'AcquiredEbTxs' has already been broadcast.
  -- The completion predicate stays true once an EB is complete, so
  -- any later batch containing one of its txs would re-trigger it
  -- without this guard; downstream consumers (e.g. 'runLeiosVoting')
  -- treat the re-notification as fatal ('AlreadyKnown' from
  -- 'addVote'). Each point carries the EB's hash and its announcer slot,
  -- so this is also the source for
  -- 'scanCompleteEbClosuresNotOlderThanSlot'.
  }
  deriving stock (forall x. InMemoryLeiosDb -> Rep InMemoryLeiosDb x)
-> (forall x. Rep InMemoryLeiosDb x -> InMemoryLeiosDb)
-> Generic InMemoryLeiosDb
forall x. Rep InMemoryLeiosDb x -> InMemoryLeiosDb
forall x. InMemoryLeiosDb -> Rep InMemoryLeiosDb x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. InMemoryLeiosDb -> Rep InMemoryLeiosDb x
from :: forall x. InMemoryLeiosDb -> Rep InMemoryLeiosDb x
$cto :: forall x. Rep InMemoryLeiosDb x -> InMemoryLeiosDb
to :: forall x. Rep InMemoryLeiosDb x -> InMemoryLeiosDb
Generic
  deriving anyclass Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo)
Proxy InMemoryLeiosDb -> String
(Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo))
-> (Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo))
-> (Proxy InMemoryLeiosDb -> String)
-> NoThunks InMemoryLeiosDb
forall a.
(Context -> a -> IO (Maybe ThunkInfo))
-> (Context -> a -> IO (Maybe ThunkInfo))
-> (Proxy a -> String)
-> NoThunks a
$cnoThunks :: Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo)
noThunks :: Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo)
$cwNoThunks :: Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo)
wNoThunks :: Context -> InMemoryLeiosDb -> IO (Maybe ThunkInfo)
$cshowTypeOf :: Proxy InMemoryLeiosDb -> String
showTypeOf :: Proxy InMemoryLeiosDb -> String
NoThunks

emptyInMemoryLeiosDb :: InMemoryLeiosDb
emptyInMemoryLeiosDb :: InMemoryLeiosDb
emptyInMemoryLeiosDb = Map TxHash (ByteString, BytesSize)
-> Map LeiosPoint BytesSize
-> Set LeiosPoint
-> Map EbHash (IntMap EbTxEntry)
-> Set LeiosPoint
-> InMemoryLeiosDb
InMemoryLeiosDb Map TxHash (ByteString, BytesSize)
forall a. Monoid a => a
mempty Map LeiosPoint BytesSize
forall a. Monoid a => a
mempty Set LeiosPoint
forall a. Monoid a => a
mempty Map EbHash (IntMap EbTxEntry)
forall a. Monoid a => a
mempty Set LeiosPoint
forall a. Monoid a => a
mempty

-- | EB transaction entry (references txs by hash, no bytes stored here)
data EbTxEntry = EbTxEntry
  { EbTxEntry -> TxHash
eteTxHash :: !TxHash
  , EbTxEntry -> BytesSize
eteTxBytesSize :: !BytesSize
  }
  deriving stock (forall x. EbTxEntry -> Rep EbTxEntry x)
-> (forall x. Rep EbTxEntry x -> EbTxEntry) -> Generic EbTxEntry
forall x. Rep EbTxEntry x -> EbTxEntry
forall x. EbTxEntry -> Rep EbTxEntry x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. EbTxEntry -> Rep EbTxEntry x
from :: forall x. EbTxEntry -> Rep EbTxEntry x
$cto :: forall x. Rep EbTxEntry x -> EbTxEntry
to :: forall x. Rep EbTxEntry x -> EbTxEntry
Generic
  deriving anyclass Context -> EbTxEntry -> IO (Maybe ThunkInfo)
Proxy EbTxEntry -> String
(Context -> EbTxEntry -> IO (Maybe ThunkInfo))
-> (Context -> EbTxEntry -> IO (Maybe ThunkInfo))
-> (Proxy EbTxEntry -> String)
-> NoThunks EbTxEntry
forall a.
(Context -> a -> IO (Maybe ThunkInfo))
-> (Context -> a -> IO (Maybe ThunkInfo))
-> (Proxy a -> String)
-> NoThunks a
$cnoThunks :: Context -> EbTxEntry -> IO (Maybe ThunkInfo)
noThunks :: Context -> EbTxEntry -> IO (Maybe ThunkInfo)
$cwNoThunks :: Context -> EbTxEntry -> IO (Maybe ThunkInfo)
wNoThunks :: Context -> EbTxEntry -> IO (Maybe ThunkInfo)
$cshowTypeOf :: Proxy EbTxEntry -> String
showTypeOf :: Proxy EbTxEntry -> String
NoThunks

-- | Create a new in-memory Leios database handle.
-- This is suitable for testing in IOSim.
newLeiosDBInMemory :: IOLike m => m (LeiosDbHandle m)
newLeiosDBInMemory :: forall (m :: * -> *). IOLike m => m (LeiosDbHandle m)
newLeiosDBInMemory = do
  stateVar <- InMemoryLeiosDb -> m (StrictTVar m InMemoryLeiosDb)
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO InMemoryLeiosDb
emptyInMemoryLeiosDb
  newLeiosDBInMemoryWith stateVar

newLeiosDBInMemoryWith :: IOLike m => StrictTVar m InMemoryLeiosDb -> m (LeiosDbHandle m)
newLeiosDBInMemoryWith :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> m (LeiosDbHandle m)
newLeiosDBInMemoryWith StrictTVar m InMemoryLeiosDb
stateVar = do
  notificationChan <- STM m (StrictTChan m LeiosEbNotification)
-> m (StrictTChan m LeiosEbNotification)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m (StrictTChan m LeiosEbNotification)
forall (m :: * -> *) a. MonadSTM m => STM m (StrictTChan m a)
newBroadcastTChan
  pure $
    LeiosDbHandle
      { -- Nothing to close: the state is a 'StrictTVar'.
        closeLeiosDbHandle = pure ()
      , subscribeEbNotifications =
          atomically (dupTChan notificationChan)
      , -- No-op for now; see 'leiosDbGarbageCollect'.
        leiosDbGarbageCollect = \SlotNo
_slotNo -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      , -- No-op for now; see 'leiosDbPromoteToImmutable'.
        leiosDbPromoteToImmutable = \[LeiosPoint]
_points -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      , -- The in-memory implementation does not track stats.
        leiosDbSampleStats = pure (LeiosDbStats 0 0 0)
      , openReader = openInMemoryReader stateVar
      , openWriter = openInMemoryWriter stateVar notificationChan
      }

-- | Reads off the 'StrictTVar'. Nothing to open or close.
openInMemoryReader :: IOLike m => StrictTVar m InMemoryLeiosDb -> m (LeiosDbReader m)
openInMemoryReader :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> m (LeiosDbReader m)
openInMemoryReader StrictTVar m InMemoryLeiosDb
stateVar =
  LeiosDbReader m -> m (LeiosDbReader m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    LeiosDbReader
      { closeReader :: m ()
closeReader = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      , lookupEbBody :: HasCallStack => EbHash -> m [(TxHash, BytesSize)]
lookupEbBody = StrictTVar m InMemoryLeiosDb -> EbHash -> m [(TxHash, BytesSize)]
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> EbHash -> m [(TxHash, BytesSize)]
imLookupEbBody StrictTVar m InMemoryLeiosDb
stateVar
      , lookupEbClosure :: HasCallStack => EbHash -> m (Maybe [(TxHash, ByteString)])
lookupEbClosure = StrictTVar m InMemoryLeiosDb
-> EbHash -> m (Maybe [(TxHash, ByteString)])
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> EbHash -> m (Maybe [(TxHash, ByteString)])
imLookupEbClosure StrictTVar m InMemoryLeiosDb
stateVar
      , batchRetrieveTxs :: HasCallStack =>
EbHash -> [Int] -> m [(Int, TxHash, Maybe ByteString)]
batchRetrieveTxs = StrictTVar m InMemoryLeiosDb
-> EbHash -> [Int] -> m [(Int, TxHash, Maybe ByteString)]
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> EbHash -> [Int] -> m [(Int, TxHash, Maybe ByteString)]
imBatchRetrieveTxs StrictTVar m InMemoryLeiosDb
stateVar
      , testScanEbPoints :: HasCallStack => m [(SlotNo, EbHash)]
testScanEbPoints = StrictTVar m InMemoryLeiosDb -> m [(SlotNo, EbHash)]
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> m [(SlotNo, EbHash)]
imScanEbPoints StrictTVar m InMemoryLeiosDb
stateVar
      , -- ThreadNet persists 'stateVar' across simulated restarts, so on
        -- restart this seeds the restored acquired-EB-closures set.
        scanCompleteEbClosuresNotOlderThanSlot :: HasCallStack => SlotNo -> m [LeiosPoint]
scanCompleteEbClosuresNotOlderThanSlot = StrictTVar m InMemoryLeiosDb -> SlotNo -> m [LeiosPoint]
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> SlotNo -> m [LeiosPoint]
imScanCompleteEbClosuresSince StrictTVar m InMemoryLeiosDb
stateVar
      }

-- | No worker and no queue: the state is a 'StrictTVar', so each write runs
-- at submission and its 'Promise' comes back already resolved. A failure is
-- captured rather than thrown here, so it surfaces at 'await' exactly like
-- the SQLite backend's.
openInMemoryWriter ::
  forall m.
  IOLike m =>
  StrictTVar m InMemoryLeiosDb ->
  StrictTChan m LeiosEbNotification ->
  m (LeiosDbWriter m)
openInMemoryWriter :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification -> m (LeiosDbWriter m)
openInMemoryWriter StrictTVar m InMemoryLeiosDb
stateVar StrictTChan m LeiosEbNotification
notificationChan =
  LeiosDbWriter m -> m (LeiosDbWriter m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    LeiosDbWriter
      { closeWriter :: m ()
closeWriter = () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      , writeEbPoint :: HasCallStack => LeiosPoint -> BytesSize -> m (Promise m ())
writeEbPoint = \LeiosPoint
point BytesSize
ebBytesSize ->
          String -> m () -> m (Promise m ())
forall a. HasCallStack => String -> m a -> m (Promise m a)
resolved (String
"WriteEbPoint " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> LeiosPoint -> String
forall a. Show a => a -> String
show LeiosPoint
point) (StrictTVar m InMemoryLeiosDb -> LeiosPoint -> BytesSize -> m ()
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> LeiosPoint -> BytesSize -> m ()
imInsertEbPoint StrictTVar m InMemoryLeiosDb
stateVar LeiosPoint
point BytesSize
ebBytesSize)
      , writeEbBody :: HasCallStack => LeiosPoint -> LeiosEb -> m (Promise m [LeiosPoint])
writeEbBody = \LeiosPoint
point LeiosEb
eb ->
          String -> m [LeiosPoint] -> m (Promise m [LeiosPoint])
forall a. HasCallStack => String -> m a -> m (Promise m a)
resolved (String
"WriteEbBody " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> LeiosPoint -> String
forall a. Show a => a -> String
show LeiosPoint
point) (StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification
-> LeiosPoint
-> LeiosEb
-> m [LeiosPoint]
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification
-> LeiosPoint
-> LeiosEb
-> m [LeiosPoint]
imInsertEbBody StrictTVar m InMemoryLeiosDb
stateVar StrictTChan m LeiosEbNotification
notificationChan LeiosPoint
point LeiosEb
eb)
      , writeTxs :: HasCallStack =>
[(TxHash, ByteString)] -> m (Promise m [LeiosPoint])
writeTxs = \[(TxHash, ByteString)]
txs ->
          String -> m [LeiosPoint] -> m (Promise m [LeiosPoint])
forall a. HasCallStack => String -> m a -> m (Promise m a)
resolved (String
"WriteTxs (" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int -> String
forall a. Show a => a -> String
show ([(TxHash, ByteString)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(TxHash, ByteString)]
txs) String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" txs)") (StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification
-> [(TxHash, ByteString)]
-> m [LeiosPoint]
forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification
-> [(TxHash, ByteString)]
-> m [LeiosPoint]
imInsertTxs StrictTVar m InMemoryLeiosDb
stateVar StrictTChan m LeiosEbNotification
notificationChan [(TxHash, ByteString)]
txs)
      }
 where
  resolved :: HasCallStack => String -> m a -> m (Promise m a)
  resolved :: forall a. HasCallStack => String -> m a -> m (Promise m a)
resolved String
job m a
action = do
    result <- m a -> m (Either LeiosDbException a)
forall e a. Exception e => m a -> m (Either e a)
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> m (Either e a)
try m a
action
    pure $ Promise $ case result of
      Right a
x -> a -> m a
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure a
x
      Left LeiosDbException
cause ->
        LeiosDbException -> m a
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (LeiosDbException -> m a) -> LeiosDbException -> m a
forall a b. (a -> b) -> a -> b
$
          LeiosDbWriteFailure -> LeiosDbException
LeiosDbWriteException
            LeiosDbWriteFailure
              { ldwfWriteJob :: String
ldwfWriteJob = String
job
              , ldwfSubmittedFrom :: String
ldwfSubmittedFrom = CallStack -> String
GHC.prettyCallStack CallStack
HasCallStack => CallStack
GHC.callStack
              , ldwfWriteFailure :: SomeException
ldwfWriteFailure = LeiosDbException -> SomeException
forall e. Exception e => e -> SomeException
toException (LeiosDbException
cause :: LeiosDbException)
              }

-- * Top-level implementations

imScanEbPoints :: IOLike m => StrictTVar m InMemoryLeiosDb -> m [(SlotNo, EbHash)]
imScanEbPoints :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> m [(SlotNo, EbHash)]
imScanEbPoints StrictTVar m InMemoryLeiosDb
stateVar = STM m [(SlotNo, EbHash)] -> m [(SlotNo, EbHash)]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [(SlotNo, EbHash)] -> m [(SlotNo, EbHash)])
-> STM m [(SlotNo, EbHash)] -> m [(SlotNo, EbHash)]
forall a b. (a -> b) -> a -> b
$ do
  state <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
  pure
    [ (pointSlotNo p, pointEbHash p)
    | p <- Map.keys (imEbPoints state)
    ]

-- | Insert an announced EB point. Idempotent: a second insert at the
-- same point keeps the first-seen size.
imInsertEbPoint :: IOLike m => StrictTVar m InMemoryLeiosDb -> LeiosPoint -> BytesSize -> m ()
imInsertEbPoint :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> LeiosPoint -> BytesSize -> m ()
imInsertEbPoint StrictTVar m InMemoryLeiosDb
stateVar LeiosPoint
point BytesSize
ebBytesSize = STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$
  StrictTVar m InMemoryLeiosDb
-> (InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m InMemoryLeiosDb
stateVar ((InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ())
-> (InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \InMemoryLeiosDb
s ->
    InMemoryLeiosDb
s{imEbPoints = Map.insertWith (\BytesSize
_ BytesSize
old -> BytesSize
old) point ebBytesSize (imEbPoints s)}

imLookupEbBody :: IOLike m => StrictTVar m InMemoryLeiosDb -> EbHash -> m [(TxHash, BytesSize)]
imLookupEbBody :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> EbHash -> m [(TxHash, BytesSize)]
imLookupEbBody StrictTVar m InMemoryLeiosDb
stateVar EbHash
ebHash = STM m [(TxHash, BytesSize)] -> m [(TxHash, BytesSize)]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [(TxHash, BytesSize)] -> m [(TxHash, BytesSize)])
-> STM m [(TxHash, BytesSize)] -> m [(TxHash, BytesSize)]
forall a b. (a -> b) -> a -> b
$ do
  state <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
  case Map.lookup ebHash (imEbBodies state) of
    Maybe (IntMap EbTxEntry)
Nothing -> [(TxHash, BytesSize)] -> STM m [(TxHash, BytesSize)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
    Just IntMap EbTxEntry
offsetMap ->
      [(TxHash, BytesSize)] -> STM m [(TxHash, BytesSize)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
        [ (EbTxEntry -> TxHash
eteTxHash EbTxEntry
e, EbTxEntry -> BytesSize
eteTxBytesSize EbTxEntry
e)
        | EbTxEntry
e <- IntMap EbTxEntry -> [EbTxEntry]
forall a. IntMap a -> [a]
IntMap.elems IntMap EbTxEntry
offsetMap
        ]

imInsertEbBody ::
  IOLike m =>
  StrictTVar m InMemoryLeiosDb ->
  StrictTChan m LeiosEbNotification ->
  LeiosPoint ->
  LeiosEb ->
  m CompletedEbs
imInsertEbBody :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification
-> LeiosPoint
-> LeiosEb
-> m [LeiosPoint]
imInsertEbBody StrictTVar m InMemoryLeiosDb
stateVar StrictTChan m LeiosEbNotification
notificationChan LeiosPoint
point LeiosEb
eb = do
  let items :: [(Int, TxHash, BytesSize)]
items = LeiosEb -> [(Int, TxHash, BytesSize)]
leiosEbBodyItems LeiosEb
eb
      ebBytesSize :: BytesSize
ebBytesSize = LeiosEb -> BytesSize
encodeLeiosEbSize LeiosEb
eb
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ([(Int, TxHash, BytesSize)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(Int, TxHash, BytesSize)]
items) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
    String -> m ()
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
String -> m a
throwLeiosDbException String
"writeEbBody: empty EB body (programmer error)"
  STM m [LeiosPoint] -> m [LeiosPoint]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [LeiosPoint] -> m [LeiosPoint])
-> STM m [LeiosPoint] -> m [LeiosPoint]
forall a b. (a -> b) -> a -> b
$ do
    let entries :: IntMap EbTxEntry
entries =
          [(Int, EbTxEntry)] -> IntMap EbTxEntry
forall a. [(Int, a)] -> IntMap a
IntMap.fromList
            [ ( Int
offset
              , EbTxEntry
                  { eteTxHash :: TxHash
eteTxHash = TxHash
txHash
                  , eteTxBytesSize :: BytesSize
eteTxBytesSize = BytesSize
size
                  }
              )
            | (Int
offset, TxHash
txHash, BytesSize
size) <- [(Int, TxHash, BytesSize)]
items
            ]
    StrictTVar m InMemoryLeiosDb
-> (InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m InMemoryLeiosDb
stateVar ((InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ())
-> (InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \InMemoryLeiosDb
s ->
      InMemoryLeiosDb
s
        { -- The tx-hash list is fully determined by the EB content
          -- hash, so a second insertion at the same hash is a no-op.
          imEbBodies =
            Map.insertWith (\IntMap EbTxEntry
_ IntMap EbTxEntry
old -> IntMap EbTxEntry
old) (pointEbHash point) entries (imEbBodies s)
        , -- Mark this point as downloaded. Other points referencing
          -- the same EB hash are unaffected; each needs its own
          -- 'writeEbBody' to be considered complete.
          imEbBodiesDownloaded =
            Set.insert point (imEbBodiesDownloaded s)
        }
    StrictTChan m LeiosEbNotification
-> LeiosEbNotification -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTChan m a -> a -> STM m ()
writeTChan StrictTChan m LeiosEbNotification
notificationChan (LeiosEbNotification -> STM m ())
-> LeiosEbNotification -> STM m ()
forall a b. (a -> b) -> a -> b
$ LeiosPoint -> BytesSize -> LeiosEbNotification
AcquiredEb LeiosPoint
point BytesSize
ebBytesSize
    -- If every tx referenced by this body is already present, the closure
    -- is complete the moment the body lands — no subsequent
    -- 'writeTxs' will fire for this point, so we must notify here.
    -- Only trigger for a novel point ('imCompletedEbs' is our idempotency
    -- guard). This mirrors what 'imInsertTxs' does when the last missing
    -- tx of an already-downloaded body arrives.
    state <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
    let allTxsPresent =
          (EbTxEntry -> Bool) -> [EbTxEntry] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
all (\EbTxEntry
e -> TxHash -> Map TxHash (ByteString, BytesSize) -> Bool
forall k a. Ord k => k -> Map k a -> Bool
Map.member (EbTxEntry -> TxHash
eteTxHash EbTxEntry
e) (InMemoryLeiosDb -> Map TxHash (ByteString, BytesSize)
imTxs InMemoryLeiosDb
state)) (IntMap EbTxEntry -> [EbTxEntry]
forall a. IntMap a -> [a]
IntMap.elems IntMap EbTxEntry
entries)
        alreadyNotified = LeiosPoint -> Set LeiosPoint -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member LeiosPoint
point (InMemoryLeiosDb -> Set LeiosPoint
imCompletedEbs InMemoryLeiosDb
state)
    if allTxsPresent && not alreadyNotified
      then do
        modifyTVar stateVar $ \InMemoryLeiosDb
s ->
          InMemoryLeiosDb
s{imCompletedEbs = Set.insert point (imCompletedEbs s)}
        writeTChan notificationChan (AcquiredEbTxs point)
        pure [point]
      else pure []

imInsertTxs ::
  IOLike m =>
  StrictTVar m InMemoryLeiosDb ->
  StrictTChan m LeiosEbNotification ->
  [(TxHash, ByteString)] ->
  m CompletedEbs
imInsertTxs :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> StrictTChan m LeiosEbNotification
-> [(TxHash, ByteString)]
-> m [LeiosPoint]
imInsertTxs StrictTVar m InMemoryLeiosDb
stateVar StrictTChan m LeiosEbNotification
notificationChan [(TxHash, ByteString)]
txs = STM m [LeiosPoint] -> m [LeiosPoint]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [LeiosPoint] -> m [LeiosPoint])
-> STM m [LeiosPoint] -> m [LeiosPoint]
forall a b. (a -> b) -> a -> b
$ do
  let insertedTxHashes :: [TxHash]
insertedTxHashes = [TxHash
txHash | (TxHash
txHash, ByteString
_) <- [(TxHash, ByteString)]
txs]
  [(TxHash, ByteString)]
-> ((TxHash, ByteString) -> STM m ()) -> STM m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [(TxHash, ByteString)]
txs (((TxHash, ByteString) -> STM m ()) -> STM m ())
-> ((TxHash, ByteString) -> STM m ()) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \(TxHash
txHash, ByteString
txBytes) -> do
    let txBytesSize :: BytesSize
txBytesSize = Int -> BytesSize
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> BytesSize) -> Int -> BytesSize
forall a b. (a -> b) -> a -> b
$ ByteString -> Int
BS.length ByteString
txBytes
    StrictTVar m InMemoryLeiosDb
-> (InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m InMemoryLeiosDb
stateVar ((InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ())
-> (InMemoryLeiosDb -> InMemoryLeiosDb) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \InMemoryLeiosDb
s ->
      if TxHash -> Map TxHash (ByteString, BytesSize) -> Bool
forall k a. Ord k => k -> Map k a -> Bool
Map.member TxHash
txHash (InMemoryLeiosDb -> Map TxHash (ByteString, BytesSize)
imTxs InMemoryLeiosDb
s)
        then InMemoryLeiosDb
s
        else InMemoryLeiosDb
s{imTxs = Map.insert txHash (txBytes, txBytesSize) (imTxs s)}
  state <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
  -- Candidates: every point whose body has been downloaded, whose
  -- hash is touched by this batch, and whose closure is now complete.
  -- A point not in 'imEbBodiesDownloaded' is skipped — its body
  -- hasn't been inserted, so it can't be complete. Two points
  -- referencing the same EB hash both light up if both have had
  -- their body inserted.
  let touchedByBatch =
        [EbHash] -> Set EbHash
forall a. Ord a => [a] -> Set a
Set.fromList
          [ EbHash
ebHash
          | (EbHash
ebHash, IntMap EbTxEntry
entries) <- Map EbHash (IntMap EbTxEntry) -> [(EbHash, IntMap EbTxEntry)]
forall k a. Map k a -> [(k, a)]
Map.toList (InMemoryLeiosDb -> Map EbHash (IntMap EbTxEntry)
imEbBodies InMemoryLeiosDb
state)
          , (EbTxEntry -> Bool) -> [EbTxEntry] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (\EbTxEntry
e -> EbTxEntry -> TxHash
eteTxHash EbTxEntry
e TxHash -> [TxHash] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [TxHash]
insertedTxHashes) (IntMap EbTxEntry -> [EbTxEntry]
forall a. IntMap a -> [a]
IntMap.elems IntMap EbTxEntry
entries)
          ]
      hashComplete EbHash
h = case EbHash -> Map EbHash (IntMap EbTxEntry) -> Maybe (IntMap EbTxEntry)
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup EbHash
h (InMemoryLeiosDb -> Map EbHash (IntMap EbTxEntry)
imEbBodies InMemoryLeiosDb
state) of
        Maybe (IntMap EbTxEntry)
Nothing -> Bool
False
        Just IntMap EbTxEntry
entries ->
          (EbTxEntry -> Bool) -> [EbTxEntry] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
all (\EbTxEntry
e -> TxHash -> Map TxHash (ByteString, BytesSize) -> Bool
forall k a. Ord k => k -> Map k a -> Bool
Map.member (EbTxEntry -> TxHash
eteTxHash EbTxEntry
e) (InMemoryLeiosDb -> Map TxHash (ByteString, BytesSize)
imTxs InMemoryLeiosDb
state)) (IntMap EbTxEntry -> [EbTxEntry]
forall a. IntMap a -> [a]
IntMap.elems IntMap EbTxEntry
entries)
      candidates =
        [ LeiosPoint
point
        | LeiosPoint
point <- Set LeiosPoint -> [LeiosPoint]
forall a. Set a -> [a]
Set.toList (InMemoryLeiosDb -> Set LeiosPoint
imEbBodiesDownloaded InMemoryLeiosDb
state)
        , EbHash -> Set EbHash -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point) Set EbHash
touchedByBatch
        , EbHash -> Bool
hashComplete (LeiosPoint -> EbHash
pointEbHash LeiosPoint
point)
        ]
      completed =
        (LeiosPoint -> Bool) -> [LeiosPoint] -> [LeiosPoint]
forall a. (a -> Bool) -> [a] -> [a]
filter
          (\LeiosPoint
p -> Bool -> Bool
not (LeiosPoint -> Set LeiosPoint -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member LeiosPoint
p (InMemoryLeiosDb -> Set LeiosPoint
imCompletedEbs InMemoryLeiosDb
state)))
          [LeiosPoint]
candidates
  modifyTVar stateVar $ \InMemoryLeiosDb
s ->
    InMemoryLeiosDb
s
      { imCompletedEbs =
          foldr Set.insert (imCompletedEbs s) completed
      }
  -- Emit a closure-completion notification for each newly-complete EB. The
  -- ChainDB subscribes to these to grow the acquired-EB-closures set it owns.
  forM_ completed $ \LeiosPoint
point ->
    StrictTChan m LeiosEbNotification
-> LeiosEbNotification -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTChan m a -> a -> STM m ()
writeTChan StrictTChan m LeiosEbNotification
notificationChan (LeiosPoint -> LeiosEbNotification
AcquiredEbTxs LeiosPoint
point)
  pure completed

-- | Implements 'scanCompleteEbClosuresNotOlderThanSlot': the already-completed EBs
-- announced no older than the given slot.
--
-- Derived from 'imCompletedEbs': an EB's greatest announcer slot is no older
-- than @sinceSlot@ exactly when at least one of its completed points has a slot
-- @>= sinceSlot@, so we collect (and dedupe) the hashes of those points. This
-- is precise across all of an EB's announcer slots, as in the SQLite backend.
imScanCompleteEbClosuresSince ::
  IOLike m => StrictTVar m InMemoryLeiosDb -> SlotNo -> m [LeiosPoint]
imScanCompleteEbClosuresSince :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb -> SlotNo -> m [LeiosPoint]
imScanCompleteEbClosuresSince StrictTVar m InMemoryLeiosDb
stateVar SlotNo
sinceSlot = STM m [LeiosPoint] -> m [LeiosPoint]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [LeiosPoint] -> m [LeiosPoint])
-> STM m [LeiosPoint] -> m [LeiosPoint]
forall a b. (a -> b) -> a -> b
$ do
  s <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
  pure
    [ p
    | p <- Set.toList (imCompletedEbs s)
    , pointSlotNo p >= sinceSlot
    ]

imBatchRetrieveTxs ::
  IOLike m => StrictTVar m InMemoryLeiosDb -> EbHash -> [Int] -> m [(Int, TxHash, Maybe ByteString)]
imBatchRetrieveTxs :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> EbHash -> [Int] -> m [(Int, TxHash, Maybe ByteString)]
imBatchRetrieveTxs StrictTVar m InMemoryLeiosDb
stateVar EbHash
ebHash [Int]
offsets = STM m [(Int, TxHash, Maybe ByteString)]
-> m [(Int, TxHash, Maybe ByteString)]
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m [(Int, TxHash, Maybe ByteString)]
 -> m [(Int, TxHash, Maybe ByteString)])
-> STM m [(Int, TxHash, Maybe ByteString)]
-> m [(Int, TxHash, Maybe ByteString)]
forall a b. (a -> b) -> a -> b
$ do
  state <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
  case Map.lookup ebHash (imEbBodies state) of
    Maybe (IntMap EbTxEntry)
Nothing -> [(Int, TxHash, Maybe ByteString)]
-> STM m [(Int, TxHash, Maybe ByteString)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
    Just IntMap EbTxEntry
offsetMap ->
      [(Int, TxHash, Maybe ByteString)]
-> STM m [(Int, TxHash, Maybe ByteString)]
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
        [ (Int
offset, EbTxEntry -> TxHash
eteTxHash EbTxEntry
entry, (ByteString, BytesSize) -> ByteString
forall a b. (a, b) -> a
fst ((ByteString, BytesSize) -> ByteString)
-> Maybe (ByteString, BytesSize) -> Maybe ByteString
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TxHash
-> Map TxHash (ByteString, BytesSize)
-> Maybe (ByteString, BytesSize)
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup (EbTxEntry -> TxHash
eteTxHash EbTxEntry
entry) (InMemoryLeiosDb -> Map TxHash (ByteString, BytesSize)
imTxs InMemoryLeiosDb
state))
        | Int
offset <- [Int]
offsets
        , Just EbTxEntry
entry <- [Int -> IntMap EbTxEntry -> Maybe EbTxEntry
forall a. Int -> IntMap a -> Maybe a
IntMap.lookup Int
offset IntMap EbTxEntry
offsetMap]
        ]

imLookupEbClosure ::
  IOLike m => StrictTVar m InMemoryLeiosDb -> EbHash -> m (Maybe [(TxHash, ByteString)])
imLookupEbClosure :: forall (m :: * -> *).
IOLike m =>
StrictTVar m InMemoryLeiosDb
-> EbHash -> m (Maybe [(TxHash, ByteString)])
imLookupEbClosure StrictTVar m InMemoryLeiosDb
stateVar EbHash
ebHash = STM m (Maybe [(TxHash, ByteString)])
-> m (Maybe [(TxHash, ByteString)])
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Maybe [(TxHash, ByteString)])
 -> m (Maybe [(TxHash, ByteString)]))
-> STM m (Maybe [(TxHash, ByteString)])
-> m (Maybe [(TxHash, ByteString)])
forall a b. (a -> b) -> a -> b
$ do
  state <- StrictTVar m InMemoryLeiosDb -> STM m InMemoryLeiosDb
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m InMemoryLeiosDb
stateVar
  case Map.lookup ebHash (imEbBodies state) of
    Maybe (IntMap EbTxEntry)
Nothing -> Maybe [(TxHash, ByteString)]
-> STM m (Maybe [(TxHash, ByteString)])
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe [(TxHash, ByteString)]
forall a. Maybe a
Nothing
    Just IntMap EbTxEntry
entries ->
      let ebTxHashes :: [TxHash]
ebTxHashes = [EbTxEntry -> TxHash
eteTxHash EbTxEntry
e | EbTxEntry
e <- IntMap EbTxEntry -> [EbTxEntry]
forall a. IntMap a -> [a]
IntMap.elems IntMap EbTxEntry
entries]
          txClosure :: [(TxHash, ByteString)]
txClosure =
            [ (TxHash
ebTxHash, ByteString
tx)
            | TxHash
ebTxHash <- [TxHash]
ebTxHashes
            , (ByteString
tx, BytesSize
_txSize) <- Maybe (ByteString, BytesSize) -> [(ByteString, BytesSize)]
forall a. Maybe a -> [a]
maybeToList (TxHash
-> Map TxHash (ByteString, BytesSize)
-> Maybe (ByteString, BytesSize)
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup TxHash
ebTxHash (InMemoryLeiosDb -> Map TxHash (ByteString, BytesSize)
imTxs InMemoryLeiosDb
state))
            ]
       in if [TxHash] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [TxHash]
ebTxHashes Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== [(TxHash, ByteString)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(TxHash, ByteString)]
txClosure
            then Maybe [(TxHash, ByteString)]
-> STM m (Maybe [(TxHash, ByteString)])
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([(TxHash, ByteString)] -> Maybe [(TxHash, ByteString)]
forall a. a -> Maybe a
Just [(TxHash, ByteString)]
txClosure)
            else Maybe [(TxHash, ByteString)]
-> STM m (Maybe [(TxHash, ByteString)])
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe [(TxHash, ByteString)]
forall a. Maybe a
Nothing