{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}

-- | Handler for the UTxO RPC @WaitForTx@ SubmitService method: stream stage
-- transitions for a set of requested transactions until every one of them
-- reaches @STAGE_CONFIRMED@.
module Cardano.Rpc.Server.Internal.UtxoRpc.WaitForTx
  ( waitForTxMethod

    -- * Internals, exported for tests
  , maxWaitForTxRefs
  , parseRequestRefs
  , waitForTxIdleTimeout
  , idleTimeLeft
  , nextIdleDeadline
  , idleDeadlineAfterStep
  , DrainedChange (..)
  , drainedBlocks
  , WaitForTxEvent (..)
  , WaitForTxState (..)
  , initialWaitForTxState
  , stepWaitForTx
  )
where

import Cardano.Api
import Cardano.Api.Consensus qualified as Consensus
import Cardano.Rpc.Proto.Api.UtxoRpc.Submit qualified as U5c
import Cardano.Rpc.Server.Internal.Error
import Cardano.Rpc.Server.Internal.Monad
import Cardano.Rpc.Server.NodeKernelAccess

import Cardano.Ledger.Core qualified as L
import Ouroboros.Consensus.Mempool.API qualified as Consensus (MempoolSnapshot (snapshotHasTx))

import RIO

import Data.ByteString qualified as BS
import Data.ProtoLens (defMessage)
import Data.Set qualified as Set
import Data.Time.Clock (DiffTime)
import Data.Traversable (mapAccumL)
import Network.GRPC.Spec
  ( GrpcError (GrpcDeadlineExceeded, GrpcInvalidArgument)
  , NextElem (NextElem, NoNextElem)
  , Proto (Proto)
  )

-- | Handle the @WaitForTx@ SubmitService RPC method.
--
-- Reports @STAGE_MEMPOOL@ when a ref enters the mempool and
-- @STAGE_CONFIRMED@ when a ref appears in a block applied after the stream
-- opened, and closes the stream once every ref has reached
-- @STAGE_CONFIRMED@.
--
-- __Threading model:__ one thread per call, with no shared state. Each step
-- is one STM wait on the chain tip and the pending refs' mempool membership,
-- bounded by 'timeout'; on a tip move the follower is drained without blocking.
--
-- __Idle timeout:__ grapesy notices a gone client only when the handler
-- sends to it, so a stream that has sent nothing for ten minutes ends with
-- @DEADLINE_EXCEEDED@. A client that still wants the outcome reopens it.
--
-- __Race contract:__ a transaction that reaches the chain between
-- submission and this call is invisible - it is no longer in the mempool
-- and will not appear in any block applied after the stream opens, so the
-- stream never reports that ref. Clients must open @WaitForTx@
-- before or concurrently with submitting the transaction.
--
-- __Reconnect gap:__ for the same reason, a confirmation that lands between
-- an idle close and the reopen is missed.
--
-- __Rollbacks are not reported:__ a ref confirmed in a block that is later
-- rolled back is not announced again; @FollowTip@'s @undo@ stream is the
-- client-side tool for that.
--
-- __Eviction is non-terminal:__ a ref that leaves the mempool without
-- on-chain inclusion produces no message; the stream keeps waiting, since
-- chain churn can legitimately re-add it later.
--
-- __Mempool delivery follows observed membership:__ a ref gets @STAGE_MEMPOOL@
-- when first observed in the mempool, and again after an observed removal and
-- re-entry. Removal and re-entry between snapshots can be missed.
--
-- Duplicate refs in the request are treated as one. An empty request
-- closes the stream immediately, with no messages sent - vacuously
-- satisfied, and without opening a follower.
--
-- Throws @INVALID_ARGUMENT@ if a ref is not a well-formed transaction id or
-- the request has more than 'maxWaitForTxRefs' refs, and
-- @DEADLINE_EXCEEDED@ when the idle timeout expires.
waitForTxMethod
  :: forall e m
   . MonadRpc e m
  => Proto U5c.WaitForTxRequest
  -- ^ Request containing the transaction ids to wait for
  -> (NextElem (Proto U5c.WaitForTxResponse) -> IO ())
  -- ^ Callback used to send each streamed response
  -> m ()
waitForTxMethod :: forall e (m :: * -> *).
MonadRpc e m =>
Proto WaitForTxRequest
-> (NextElem (Proto WaitForTxResponse) -> IO ()) -> m ()
waitForTxMethod Proto WaitForTxRequest
request NextElem (Proto WaitForTxResponse) -> IO ()
send = do
  refs <- [ByteString] -> m (Set TxId)
forall (m :: * -> *). MonadIO m => [ByteString] -> m (Set TxId)
parseRequestRefs ([ByteString] -> m (Set TxId)) -> [ByteString] -> m (Set TxId)
forall a b. (a -> b) -> a -> b
$ Proto WaitForTxRequest
request Proto WaitForTxRequest
-> Getting [ByteString] (Proto WaitForTxRequest) [ByteString]
-> [ByteString]
forall s a. s -> Getting a s a -> a
^. Getting [ByteString] (Proto WaitForTxRequest) [ByteString]
forall (f :: * -> *) s a.
(Functor f, HasField s "ref" a) =>
LensLike' f s a
U5c.ref
  case initialWaitForTxState refs of
    Maybe WaitForTxState
Nothing -> IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ NextElem (Proto WaitForTxResponse) -> IO ()
send NextElem (Proto WaitForTxResponse)
forall a. NextElem a
NoNextElem
    Just WaitForTxState
state -> do
      nodeKernelAccess <- m NodeKernelAccess
forall e (m :: * -> *). MonadRpc e m => m NodeKernelAccess
grabNodeKernelAccess
      withBlockFollower nodeKernelAccess $ \ChainFollower BlockInMode
follower -> do
        tip <- NodeKernelAccess -> ChainFollower BlockInMode -> m ChainPoint
forall (m :: * -> *) a.
MonadIO m =>
NodeKernelAccess -> ChainFollower a -> m ChainPoint
positionFollowerAtCurrentTip NodeKernelAccess
nodeKernelAccess ChainFollower BlockInMode
follower
        now <- getMonotonicDiffTime
        waitLoop nodeKernelAccess follower state tip (now + waitForTxIdleTimeout)
 where
  -- Wait for the next node change before the idle deadline, step the core on
  -- it and send its responses, until the core closes the stream.
  waitLoop
    :: NodeKernelAccess
    -> ChainFollower BlockInMode
    -> WaitForTxState
    -> ChainPoint
    -> DiffTime
    -> m ()
  waitLoop :: NodeKernelAccess
-> ChainFollower BlockInMode
-> WaitForTxState
-> ChainPoint
-> DiffTime
-> m ()
waitLoop NodeKernelAccess
nodeKernelAccess ChainFollower BlockInMode
follower WaitForTxState
state ChainPoint
lastTip DiffTime
deadline = do
    now <- m DiffTime
forall (m :: * -> *). MonadIO m => m DiffTime
getMonotonicDiffTime
    (change, expired) <- case idleTimeLeft deadline now of
      Maybe Int
Nothing -> (,Bool
True) (NodeChange -> (NodeChange, Bool))
-> m NodeChange -> m (NodeChange, Bool)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m NodeChange
readyChangeOrIdleTimeout
      Just Int
timeLeft ->
        Int -> m NodeChange -> m (Maybe NodeChange)
forall (m :: * -> *) a.
MonadUnliftIO m =>
Int -> m a -> m (Maybe a)
timeout Int
timeLeft (STM NodeChange -> m NodeChange
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically STM NodeChange
nextChange) m (Maybe NodeChange)
-> (Maybe NodeChange -> m (NodeChange, Bool))
-> m (NodeChange, Bool)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          Maybe NodeChange
Nothing -> (,Bool
True) (NodeChange -> (NodeChange, Bool))
-> m NodeChange -> m (NodeChange, Bool)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m NodeChange
readyChangeOrIdleTimeout
          Just NodeChange
waited -> (NodeChange, Bool) -> m (NodeChange, Bool)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (NodeChange
waited, Bool
False)
    (event, tip) <- case change of
      TipMoved ChainPoint
newTip -> do
        blocks <- ChainFollower BlockInMode -> Set TxId -> m [Set TxId]
forall (m :: * -> *).
MonadIO m =>
ChainFollower BlockInMode -> Set TxId -> m [Set TxId]
drainFollower ChainFollower BlockInMode
follower (Set TxId -> m [Set TxId]) -> Set TxId -> m [Set TxId]
forall a b. (a -> b) -> a -> b
$ WaitForTxState -> Set TxId
pendingRefs WaitForTxState
state
        pure (TipBlocks blocks, newTip)
      MempoolChanged Set TxId
inMempool -> (WaitForTxEvent, ChainPoint) -> m (WaitForTxEvent, ChainPoint)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Set TxId -> WaitForTxEvent
MempoolMembership Set TxId
inMempool, ChainPoint
lastTip)
    let (next, responses) = stepWaitForTx state event
    liftIO $ mapM_ (send . NextElem) responses
    sentAt <- getMonotonicDiffTime
    case next of
      Maybe WaitForTxState
Nothing -> IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ NextElem (Proto WaitForTxResponse) -> IO ()
send NextElem (Proto WaitForTxResponse)
forall a. NextElem a
NoNextElem
      Just WaitForTxState
state' ->
        m () -> (DiffTime -> m ()) -> Maybe DiffTime -> m ()
forall b a. b -> (a -> b) -> Maybe a -> b
maybe m ()
forall a. m a
throwIdleTimeout (NodeKernelAccess
-> ChainFollower BlockInMode
-> WaitForTxState
-> ChainPoint
-> DiffTime
-> m ()
waitLoop NodeKernelAccess
nodeKernelAccess ChainFollower BlockInMode
follower WaitForTxState
state' ChainPoint
tip) (Maybe DiffTime -> m ()) -> Maybe DiffTime -> m ()
forall a b. (a -> b) -> a -> b
$
          Bool
-> DiffTime
-> DiffTime
-> [Proto WaitForTxResponse]
-> Maybe DiffTime
idleDeadlineAfterStep Bool
expired DiffTime
sentAt DiffTime
deadline [Proto WaitForTxResponse]
responses
   where
    nextChange :: STM NodeChange
    nextChange :: STM NodeChange
nextChange = NodeKernelAccess -> WaitForTxState -> ChainPoint -> STM NodeChange
nextNodeChange NodeKernelAccess
nodeKernelAccess WaitForTxState
state ChainPoint
lastTip

    -- Past the idle deadline, still take a change that is already waiting,
    -- and only then end the stream.
    readyChangeOrIdleTimeout :: m NodeChange
    readyChangeOrIdleTimeout :: m NodeChange
readyChangeOrIdleTimeout = m NodeChange
-> (NodeChange -> m NodeChange) -> Maybe NodeChange -> m NodeChange
forall b a. b -> (a -> b) -> Maybe a -> b
maybe m NodeChange
forall a. m a
throwIdleTimeout NodeChange -> m NodeChange
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe NodeChange -> m NodeChange)
-> m (Maybe NodeChange) -> m NodeChange
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< STM (Maybe NodeChange) -> m (Maybe NodeChange)
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM NodeChange -> STM (Maybe NodeChange)
forall (f :: * -> *) a. Alternative f => f a -> f (Maybe a)
optional STM NodeChange
nextChange)

  throwIdleTimeout :: m a
  throwIdleTimeout :: forall a. m a
throwIdleTimeout =
    GrpcError -> Text -> m a
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage
      GrpcError
GrpcDeadlineExceeded
      Text
"no WaitForTx update for 10 minutes, reopen the stream to keep waiting"

-- | The most refs one @WaitForTx@ request may carry. Each pending ref is
-- checked against the mempool on every wake-up.
maxWaitForTxRefs :: Int
maxWaitForTxRefs :: Int
maxWaitForTxRefs = Int
500

-- | Parse the request refs into a set of transaction ids.
-- Throws @INVALID_ARGUMENT@ if there are more than 'maxWaitForTxRefs' refs,
-- duplicates included, or if a ref is not a well-formed transaction id.
parseRequestRefs :: MonadIO m => [ByteString] -> m (Set TxId)
parseRequestRefs :: forall (m :: * -> *). MonadIO m => [ByteString] -> m (Set TxId)
parseRequestRefs [ByteString]
refs = do
  let refCount :: Int
refCount = [ByteString] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [ByteString]
refs
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
refCount Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
maxWaitForTxRefs) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
    GrpcError -> Text -> m ()
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage GrpcError
GrpcInvalidArgument (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$
      Text
"too many transaction refs: "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
tshow Int
refCount
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
", maximum "
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
tshow Int
maxWaitForTxRefs
        Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"; batch your requests"
  [TxId] -> Set TxId
forall a. Ord a => [a] -> Set a
Set.fromList ([TxId] -> Set TxId) -> m [TxId] -> m (Set TxId)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (ByteString -> m TxId) -> [ByteString] -> m [TxId]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse ByteString -> m TxId
forall (m :: * -> *). MonadIO m => ByteString -> m TxId
parseRef [ByteString]
refs

-- | How long a @WaitForTx@ stream may go without sending a message before it
-- ends with @DEADLINE_EXCEEDED@: ten minutes.
waitForTxIdleTimeout :: DiffTime
waitForTxIdleTimeout :: DiffTime
waitForTxIdleTimeout = DiffTime
10 DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
* DiffTime
60

-- | The microseconds left until the idle deadline, rounded up so that they
-- never become a zero wait, or 'Nothing' once the deadline has passed.
idleTimeLeft
  :: DiffTime
  -- ^ The idle deadline, on the monotonic clock
  -> DiffTime
  -- ^ Monotonic time now
  -> Maybe Int
idleTimeLeft :: DiffTime -> DiffTime -> Maybe Int
idleTimeLeft DiffTime
deadline DiffTime
now
  | DiffTime
now DiffTime -> DiffTime -> Bool
forall a. Ord a => a -> a -> Bool
>= DiffTime
deadline = Maybe Int
forall a. Maybe a
Nothing
  | Bool
otherwise = Int -> Maybe Int
forall a. a -> Maybe a
Just (Int -> Maybe Int) -> (DiffTime -> Int) -> DiffTime -> Maybe Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. DiffTime -> Int
toMicros (DiffTime -> Maybe Int) -> DiffTime -> Maybe Int
forall a b. (a -> b) -> a -> b
$ DiffTime
deadline DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
- DiffTime
now

-- | The idle deadline after a step: one 'waitForTxIdleTimeout' from now if the
-- step sent a response, unchanged otherwise.
nextIdleDeadline
  :: DiffTime
  -- ^ Monotonic time now, after sending
  -> DiffTime
  -- ^ The current idle deadline
  -> [Proto U5c.WaitForTxResponse]
  -- ^ The responses the step sent
  -> DiffTime
nextIdleDeadline :: DiffTime -> DiffTime -> [Proto WaitForTxResponse] -> DiffTime
nextIdleDeadline DiffTime
now DiffTime
deadline [Proto WaitForTxResponse]
responses
  | [Proto WaitForTxResponse] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Proto WaitForTxResponse]
responses = DiffTime
deadline
  | Bool
otherwise = DiffTime
now DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
+ DiffTime
waitForTxIdleTimeout

-- | The idle deadline after a step, or 'Nothing' if the stream must end with
-- @DEADLINE_EXCEEDED@: a change taken past the deadline that sent no response.
idleDeadlineAfterStep
  :: Bool
  -- ^ Whether the change was taken past the idle deadline
  -> DiffTime
  -- ^ Monotonic time now, after sending
  -> DiffTime
  -- ^ The current idle deadline
  -> [Proto U5c.WaitForTxResponse]
  -- ^ The responses the step sent
  -> Maybe DiffTime
idleDeadlineAfterStep :: Bool
-> DiffTime
-> DiffTime
-> [Proto WaitForTxResponse]
-> Maybe DiffTime
idleDeadlineAfterStep Bool
expired DiffTime
now DiffTime
deadline [Proto WaitForTxResponse]
responses
  | Bool
expired Bool -> Bool -> Bool
&& [Proto WaitForTxResponse] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Proto WaitForTxResponse]
responses = Maybe DiffTime
forall a. Maybe a
Nothing
  | Bool
otherwise = DiffTime -> Maybe DiffTime
forall a. a -> Maybe a
Just (DiffTime -> Maybe DiffTime) -> DiffTime -> Maybe DiffTime
forall a b. (a -> b) -> a -> b
$ DiffTime -> DiffTime -> [Proto WaitForTxResponse] -> DiffTime
nextIdleDeadline DiffTime
now DiffTime
deadline [Proto WaitForTxResponse]
responses

-- | Microseconds for 'timeout', rounded up.
toMicros :: DiffTime -> Int
toMicros :: DiffTime -> Int
toMicros = DiffTime -> Int
forall b. Integral b => DiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (DiffTime -> Int) -> (DiffTime -> DiffTime) -> DiffTime -> Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
* DiffTime
1000000)

-- | The monotonic clock, in seconds since some fixed point.
getMonotonicDiffTime :: MonadIO m => m DiffTime
getMonotonicDiffTime :: forall (m :: * -> *). MonadIO m => m DiffTime
getMonotonicDiffTime = Double -> DiffTime
forall a b. (Real a, Fractional b) => a -> b
realToFrac (Double -> DiffTime) -> m Double -> m DiffTime
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m Double
forall (m :: * -> *). MonadIO m => m Double
getMonotonicTime

-- | A change on the node that 'waitForTxMethod' wakes up for.
data NodeChange
  = -- | The chain tip moved to the given point.
    TipMoved ChainPoint
  | -- | The pending refs now in the mempool, which differ from those
    -- already announced.
    MempoolChanged (Set TxId)

-- | Wait in STM for the chain tip to move away from the last one seen or,
-- failing that, for the pending refs in the mempool to differ from those
-- already announced.
nextNodeChange
  :: NodeKernelAccess
  -> WaitForTxState
  -> ChainPoint
  -- ^ The last tip seen, which the follower has been drained up to
  -> STM NodeChange
nextNodeChange :: NodeKernelAccess -> WaitForTxState -> ChainPoint -> STM NodeChange
nextNodeChange NodeKernelAccess
nodeKernelAccess WaitForTxState{Set TxId
pendingRefs :: WaitForTxState -> Set TxId
pendingRefs :: Set TxId
pendingRefs, Set TxId
mempoolNotifiedRefs :: Set TxId
mempoolNotifiedRefs :: WaitForTxState -> Set TxId
mempoolNotifiedRefs} ChainPoint
lastTip =
  STM NodeChange
tipMoved STM NodeChange -> STM NodeChange -> STM NodeChange
forall a. STM a -> STM a -> STM a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> STM NodeChange
mempoolChanged
 where
  tipMoved :: STM NodeChange
  tipMoved :: STM NodeChange
tipMoved = do
    tip <- NodeKernelAccess -> STM ChainPoint
readChainTipPoint NodeKernelAccess
nodeKernelAccess
    checkSTM $ tip /= lastTip
    pure $ TipMoved tip

  mempoolChanged :: STM NodeChange
  mempoolChanged :: STM NodeChange
mempoolChanged = do
    -- Ids in the ledger's current era, the one the mempool stores txs in. Consensus compares
    -- ids by hash only, so the brief ledger/mempool era gap at a hard fork is harmless.
    AnyCardanoEra era <- NodeKernelAccess -> STM AnyCardanoEra
readCurrentEra NodeKernelAccess
nodeKernelAccess
    snapshot <- readMempoolSnapshot nodeKernelAccess
    let inMempool =
          (TxId -> Bool) -> Set TxId -> Set TxId
forall a. (a -> Bool) -> Set a -> Set a
Set.filter
            (MempoolSnapshot (CardanoBlock StandardCrypto)
-> GenTxId (CardanoBlock StandardCrypto) -> Bool
forall blk. MempoolSnapshot blk -> GenTxId blk -> Bool
Consensus.snapshotHasTx MempoolSnapshot (CardanoBlock StandardCrypto)
snapshot (GenTxId (CardanoBlock StandardCrypto) -> Bool)
-> (TxId -> GenTxId (CardanoBlock StandardCrypto)) -> TxId -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TxIdInMode -> GenTxId (CardanoBlock StandardCrypto)
forall block.
(CardanoBlock StandardCrypto ~ block) =>
TxIdInMode -> TxId (GenTx block)
Consensus.toConsensusTxId (TxIdInMode -> GenTxId (CardanoBlock StandardCrypto))
-> (TxId -> TxIdInMode)
-> TxId
-> GenTxId (CardanoBlock StandardCrypto)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. CardanoEra era -> TxId -> TxIdInMode
forall era. CardanoEra era -> TxId -> TxIdInMode
Consensus.TxIdInMode CardanoEra era
era)
            Set TxId
pendingRefs
    checkSTM $ inMempool /= mempoolNotifiedRefs
    pure $ MempoolChanged inMempool

-- | Take every chain update the follower has ready, without blocking, and
-- return the pending refs in each applied block that the same drain did not
-- roll back, oldest block first.
drainFollower
  :: forall m
   . MonadIO m
  => ChainFollower BlockInMode
  -> Set TxId
  -- ^ The pending refs
  -> m [Set TxId]
drainFollower :: forall (m :: * -> *).
MonadIO m =>
ChainFollower BlockInMode -> Set TxId -> m [Set TxId]
drainFollower ChainFollower{forall (m :: * -> *).
MonadIO m =>
m (Maybe (ChainChange BlockInMode))
tryNextChange :: forall (m :: * -> *).
MonadIO m =>
m (Maybe (ChainChange BlockInMode))
tryNextChange :: forall a.
ChainFollower a
-> forall (m :: * -> *). MonadIO m => m (Maybe (ChainChange a))
tryNextChange} Set TxId
pendingRefs = [DrainedChange] -> [Set TxId]
drainedBlocks ([DrainedChange] -> [Set TxId])
-> m [DrainedChange] -> m [Set TxId]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [DrainedChange] -> m [DrainedChange]
go []
 where
  go :: [DrainedChange] -> m [DrainedChange]
  go :: [DrainedChange] -> m [DrainedChange]
go [DrainedChange]
changes =
    m (Maybe (ChainChange BlockInMode))
forall (m :: * -> *).
MonadIO m =>
m (Maybe (ChainChange BlockInMode))
tryNextChange m (Maybe (ChainChange BlockInMode))
-> (Maybe (ChainChange BlockInMode) -> m [DrainedChange])
-> m [DrainedChange]
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Maybe (ChainChange BlockInMode)
Nothing -> [DrainedChange] -> m [DrainedChange]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([DrainedChange] -> m [DrainedChange])
-> [DrainedChange] -> m [DrainedChange]
forall a b. (a -> b) -> a -> b
$ [DrainedChange] -> [DrainedChange]
forall a. [a] -> [a]
reverse [DrainedChange]
changes
      Just (ChainApply blockInMode :: BlockInMode
blockInMode@(BlockInMode CardanoEra era
_ Block era
block)) -> do
        let BlockHeader SlotNo
slot Hash BlockHeader
headerHash BlockNo
_ = Block era -> BlockHeader
forall era. Block era -> BlockHeader
getBlockHeader Block era
block
            inBlock :: Set TxId
inBlock = Set TxId
pendingRefs Set TxId -> Set TxId -> Set TxId
forall a. Ord a => Set a -> Set a -> Set a
`Set.intersection` BlockInMode -> Set TxId
blockTxIds BlockInMode
blockInMode
        [DrainedChange] -> m [DrainedChange]
go ([DrainedChange] -> m [DrainedChange])
-> [DrainedChange] -> m [DrainedChange]
forall a b. (a -> b) -> a -> b
$ ChainPoint -> Set TxId -> DrainedChange
DrainedBlock (SlotNo -> Hash BlockHeader -> ChainPoint
ChainPoint SlotNo
slot Hash BlockHeader
headerHash) Set TxId
inBlock DrainedChange -> [DrainedChange] -> [DrainedChange]
forall a. a -> [a] -> [a]
: [DrainedChange]
changes
      Just (ChainRollBack ChainPoint
point) -> [DrainedChange] -> m [DrainedChange]
go ([DrainedChange] -> m [DrainedChange])
-> [DrainedChange] -> m [DrainedChange]
forall a b. (a -> b) -> a -> b
$ ChainPoint -> DrainedChange
DrainedRollBack ChainPoint
point DrainedChange -> [DrainedChange] -> [DrainedChange]
forall a. a -> [a] -> [a]
: [DrainedChange]
changes

-- | A chain update taken by 'drainFollower', reduced to what @WaitForTx@
-- needs.
data DrainedChange
  = -- | An applied block's point and the pending refs in it.
    DrainedBlock ChainPoint (Set TxId)
  | -- | A rollback to the given point.
    DrainedRollBack ChainPoint
  deriving Int -> DrainedChange -> ShowS
[DrainedChange] -> ShowS
DrainedChange -> String
(Int -> DrainedChange -> ShowS)
-> (DrainedChange -> String)
-> ([DrainedChange] -> ShowS)
-> Show DrainedChange
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> DrainedChange -> ShowS
showsPrec :: Int -> DrainedChange -> ShowS
$cshow :: DrainedChange -> String
show :: DrainedChange -> String
$cshowList :: [DrainedChange] -> ShowS
showList :: [DrainedChange] -> ShowS
Show

-- | The refs of each drained block that no later rollback in the same drain
-- undid, oldest block first. A rollback to a point before every drained
-- block undoes them all.
drainedBlocks :: [DrainedChange] -> [Set TxId]
drainedBlocks :: [DrainedChange] -> [Set TxId]
drainedBlocks = [Set TxId] -> [Set TxId]
forall a. [a] -> [a]
reverse ([Set TxId] -> [Set TxId])
-> ([DrainedChange] -> [Set TxId]) -> [DrainedChange] -> [Set TxId]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ((ChainPoint, Set TxId) -> Set TxId)
-> [(ChainPoint, Set TxId)] -> [Set TxId]
forall a b. (a -> b) -> [a] -> [b]
map (ChainPoint, Set TxId) -> Set TxId
forall a b. (a, b) -> b
snd ([(ChainPoint, Set TxId)] -> [Set TxId])
-> ([DrainedChange] -> [(ChainPoint, Set TxId)])
-> [DrainedChange]
-> [Set TxId]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ([(ChainPoint, Set TxId)]
 -> DrainedChange -> [(ChainPoint, Set TxId)])
-> [(ChainPoint, Set TxId)]
-> [DrainedChange]
-> [(ChainPoint, Set TxId)]
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' [(ChainPoint, Set TxId)]
-> DrainedChange -> [(ChainPoint, Set TxId)]
step []
 where
  -- Blocks newest first, so a rollback drops the prefix after its target.
  step :: [(ChainPoint, Set TxId)] -> DrainedChange -> [(ChainPoint, Set TxId)]
  step :: [(ChainPoint, Set TxId)]
-> DrainedChange -> [(ChainPoint, Set TxId)]
step [(ChainPoint, Set TxId)]
blocks = \case
    DrainedBlock ChainPoint
point Set TxId
refs -> (ChainPoint
point, Set TxId
refs) (ChainPoint, Set TxId)
-> [(ChainPoint, Set TxId)] -> [(ChainPoint, Set TxId)]
forall a. a -> [a] -> [a]
: [(ChainPoint, Set TxId)]
blocks
    DrainedRollBack ChainPoint
point -> ((ChainPoint, Set TxId) -> Bool)
-> [(ChainPoint, Set TxId)] -> [(ChainPoint, Set TxId)]
forall a. (a -> Bool) -> [a] -> [a]
dropWhile ((ChainPoint -> ChainPoint -> Bool
forall a. Eq a => a -> a -> Bool
/= ChainPoint
point) (ChainPoint -> Bool)
-> ((ChainPoint, Set TxId) -> ChainPoint)
-> (ChainPoint, Set TxId)
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (ChainPoint, Set TxId) -> ChainPoint
forall a b. (a, b) -> a
fst) [(ChainPoint, Set TxId)]
blocks

-- | An event 'stepWaitForTx' handles.
data WaitForTxEvent
  = -- | The chain tip moved: the transaction ids in the blocks applied since
    -- the previous tip move, one set per block, oldest first. Ids that are
    -- not pending may be left out.
    TipBlocks [Set TxId]
  | -- | The pending refs now in the mempool, after that set changed.
    MempoolMembership (Set TxId)
  deriving (WaitForTxEvent -> WaitForTxEvent -> Bool
(WaitForTxEvent -> WaitForTxEvent -> Bool)
-> (WaitForTxEvent -> WaitForTxEvent -> Bool) -> Eq WaitForTxEvent
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: WaitForTxEvent -> WaitForTxEvent -> Bool
== :: WaitForTxEvent -> WaitForTxEvent -> Bool
$c/= :: WaitForTxEvent -> WaitForTxEvent -> Bool
/= :: WaitForTxEvent -> WaitForTxEvent -> Bool
Eq, Int -> WaitForTxEvent -> ShowS
[WaitForTxEvent] -> ShowS
WaitForTxEvent -> String
(Int -> WaitForTxEvent -> ShowS)
-> (WaitForTxEvent -> String)
-> ([WaitForTxEvent] -> ShowS)
-> Show WaitForTxEvent
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> WaitForTxEvent -> ShowS
showsPrec :: Int -> WaitForTxEvent -> ShowS
$cshow :: WaitForTxEvent -> String
show :: WaitForTxEvent -> String
$cshowList :: [WaitForTxEvent] -> ShowS
showList :: [WaitForTxEvent] -> ShowS
Show)

-- | Every transaction id confirmed in a block. Byron blocks contribute no
-- ids ('getBlockTxs' returns @[]@ for them: Byron transactions predate
-- cardano-api's 'Tx' entirely, and cannot occur on any network still
-- running today).
--
-- Computed straight off the ledger tx via 'L.txIdTx', the same route
-- 'Cardano.Rpc.Server.Internal.UtxoRpc.Type.Mempool.txInModeToTxInMempool'
-- and 'Cardano.Rpc.Server.Internal.UtxoRpc.Submit.submitTxMethod' use -
-- 'getTxId'\/'getTxBody' would work too, but 'getTxBody' is deprecated in
-- favour of the experimental API.
blockTxIds :: BlockInMode -> Set TxId
blockTxIds :: BlockInMode -> Set TxId
blockTxIds (BlockInMode CardanoEra era
_ Block era
block) = case Block era
block of
  ByronBlock{} -> Set TxId
forall a. Set a
Set.empty
  ShelleyBlock ShelleyBasedEra era
sbe ShelleyBlock (ConsensusProtocol era) (ShelleyLedgerEra era)
_ ->
    ShelleyBasedEra era
-> (ShelleyBasedEraConstraints era => Set TxId) -> Set TxId
forall era a.
ShelleyBasedEra era -> (ShelleyBasedEraConstraints era => a) -> a
shelleyBasedEraConstraints ShelleyBasedEra era
sbe ((ShelleyBasedEraConstraints era => Set TxId) -> Set TxId)
-> (ShelleyBasedEraConstraints era => Set TxId) -> Set TxId
forall a b. (a -> b) -> a -> b
$
      [TxId] -> Set TxId
forall a. Ord a => [a] -> Set a
Set.fromList
        [TxId -> TxId
fromShelleyTxId (Tx TopTx (ShelleyLedgerEra era) -> TxId
forall era (l :: TxLevel). EraTx era => Tx l era -> TxId
L.txIdTx Tx TopTx (ShelleyLedgerEra era)
ledgerTx) | ShelleyTx ShelleyBasedEra era
_ Tx TopTx (ShelleyLedgerEra era)
ledgerTx <- Block era -> [Tx era]
forall era. Block era -> [Tx era]
getBlockTxs Block era
block]

-- | Parse a request ref into a transaction id.
-- Throws @INVALID_ARGUMENT@ if the bytes are not a well-formed transaction id.
parseRef :: forall m. MonadIO m => ByteString -> m TxId
parseRef :: forall (m :: * -> *). MonadIO m => ByteString -> m TxId
parseRef ByteString
bytes =
  (SerialiseAsRawBytesError -> m TxId)
-> (TxId -> m TxId)
-> Either SerialiseAsRawBytesError TxId
-> m TxId
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (m TxId -> SerialiseAsRawBytesError -> m TxId
forall a b. a -> b -> a
const m TxId
throwInvalidRef) TxId -> m TxId
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either SerialiseAsRawBytesError TxId -> m TxId)
-> Either SerialiseAsRawBytesError TxId -> m TxId
forall a b. (a -> b) -> a -> b
$ AsType TxId -> ByteString -> Either SerialiseAsRawBytesError TxId
forall a.
SerialiseAsRawBytes a =>
AsType a -> ByteString -> Either SerialiseAsRawBytesError a
deserialiseFromRawBytes AsType TxId
AsTxId ByteString
bytes
 where
  throwInvalidRef :: m TxId
  throwInvalidRef :: m TxId
throwInvalidRef =
    GrpcError -> Text -> m TxId
forall (m :: * -> *) a. MonadIO m => GrpcError -> Text -> m a
throwGrpcErrorWithMessage GrpcError
GrpcInvalidArgument (Text -> m TxId) -> Text -> m TxId
forall a b. (a -> b) -> a -> b
$
      Text
"invalid transaction id (" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall a. Show a => a -> Text
tshow (ByteString -> Int
BS.length ByteString
bytes) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" bytes)"

-- | The per-ref state of a @WaitForTx@ stream: refs still awaiting
-- confirmation, and the refs already announced as in the mempool and still
-- in it (so an evicted-then-re-added ref is announced again).
data WaitForTxState = WaitForTxState
  { WaitForTxState -> Set TxId
pendingRefs :: !(Set TxId)
  , WaitForTxState -> Set TxId
mempoolNotifiedRefs :: !(Set TxId)
  }
  deriving (WaitForTxState -> WaitForTxState -> Bool
(WaitForTxState -> WaitForTxState -> Bool)
-> (WaitForTxState -> WaitForTxState -> Bool) -> Eq WaitForTxState
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: WaitForTxState -> WaitForTxState -> Bool
== :: WaitForTxState -> WaitForTxState -> Bool
$c/= :: WaitForTxState -> WaitForTxState -> Bool
/= :: WaitForTxState -> WaitForTxState -> Bool
Eq, Int -> WaitForTxState -> ShowS
[WaitForTxState] -> ShowS
WaitForTxState -> String
(Int -> WaitForTxState -> ShowS)
-> (WaitForTxState -> String)
-> ([WaitForTxState] -> ShowS)
-> Show WaitForTxState
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> WaitForTxState -> ShowS
showsPrec :: Int -> WaitForTxState -> ShowS
$cshow :: WaitForTxState -> String
show :: WaitForTxState -> String
$cshowList :: [WaitForTxState] -> ShowS
showList :: [WaitForTxState] -> ShowS
Show)

-- | The state of a new stream for the given refs, already deduplicated, or
-- 'Nothing' if there are none and the stream closes at once.
initialWaitForTxState :: Set TxId -> Maybe WaitForTxState
initialWaitForTxState :: Set TxId -> Maybe WaitForTxState
initialWaitForTxState Set TxId
refs =
  WaitForTxState -> Maybe WaitForTxState
whilePending WaitForTxState{pendingRefs :: Set TxId
pendingRefs = Set TxId
refs, mempoolNotifiedRefs :: Set TxId
mempoolNotifiedRefs = Set TxId
forall a. Set a
Set.empty}

-- | One step of the @WaitForTx@ state machine: the next state, or 'Nothing'
-- once every ref is confirmed and the stream closes, and the responses to
-- send, in order. Confirmations follow block order.
stepWaitForTx
  :: WaitForTxState
  -> WaitForTxEvent
  -> (Maybe WaitForTxState, [Proto U5c.WaitForTxResponse])
stepWaitForTx :: WaitForTxState
-> WaitForTxEvent
-> (Maybe WaitForTxState, [Proto WaitForTxResponse])
stepWaitForTx WaitForTxState
state =
  (WaitForTxState -> Maybe WaitForTxState)
-> (WaitForTxState, [Proto WaitForTxResponse])
-> (Maybe WaitForTxState, [Proto WaitForTxResponse])
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first WaitForTxState -> Maybe WaitForTxState
whilePending ((WaitForTxState, [Proto WaitForTxResponse])
 -> (Maybe WaitForTxState, [Proto WaitForTxResponse]))
-> (WaitForTxEvent -> (WaitForTxState, [Proto WaitForTxResponse]))
-> WaitForTxEvent
-> (Maybe WaitForTxState, [Proto WaitForTxResponse])
forall b c a. (b -> c) -> (a -> b) -> a -> c
. \case
    TipBlocks [Set TxId]
blocks -> [[Proto WaitForTxResponse]] -> [Proto WaitForTxResponse]
forall (t :: * -> *) a. Foldable t => t [a] -> [a]
concat ([[Proto WaitForTxResponse]] -> [Proto WaitForTxResponse])
-> (WaitForTxState, [[Proto WaitForTxResponse]])
-> (WaitForTxState, [Proto WaitForTxResponse])
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (WaitForTxState
 -> Set TxId -> (WaitForTxState, [Proto WaitForTxResponse]))
-> WaitForTxState
-> [Set TxId]
-> (WaitForTxState, [[Proto WaitForTxResponse]])
forall (t :: * -> *) s a b.
Traversable t =>
(s -> a -> (s, b)) -> s -> t a -> (s, t b)
mapAccumL WaitForTxState
-> Set TxId -> (WaitForTxState, [Proto WaitForTxResponse])
confirmBlock WaitForTxState
state [Set TxId]
blocks
    MempoolMembership Set TxId
inMempool -> WaitForTxState
-> Set TxId -> (WaitForTxState, [Proto WaitForTxResponse])
announceMempool WaitForTxState
state Set TxId
inMempool

-- | Confirm the pending refs in one block: @STAGE_CONFIRMED@ for each, and
-- each leaves both sets.
confirmBlock
  :: WaitForTxState
  -> Set TxId
  -> (WaitForTxState, [Proto U5c.WaitForTxResponse])
confirmBlock :: WaitForTxState
-> Set TxId -> (WaitForTxState, [Proto WaitForTxResponse])
confirmBlock WaitForTxState{Set TxId
pendingRefs :: WaitForTxState -> Set TxId
pendingRefs :: Set TxId
pendingRefs, Set TxId
mempoolNotifiedRefs :: WaitForTxState -> Set TxId
mempoolNotifiedRefs :: Set TxId
mempoolNotifiedRefs} Set TxId
block =
  ( WaitForTxState
      { pendingRefs :: Set TxId
pendingRefs = Set TxId
pendingRefs Set TxId -> Set TxId -> Set TxId
forall a. Ord a => Set a -> Set a -> Set a
`Set.difference` Set TxId
confirmed
      , mempoolNotifiedRefs :: Set TxId
mempoolNotifiedRefs = Set TxId
mempoolNotifiedRefs Set TxId -> Set TxId -> Set TxId
forall a. Ord a => Set a -> Set a -> Set a
`Set.difference` Set TxId
confirmed
      }
  , [ Proto WaitForTxResponse
forall msg. Message msg => msg
defMessage
        Proto WaitForTxResponse
-> (Proto WaitForTxResponse -> Proto WaitForTxResponse)
-> Proto WaitForTxResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto WaitForTxResponse) ByteString
forall (f :: * -> *) s a.
(Functor f, HasField s "ref" a) =>
LensLike' f s a
U5c.ref LensLike' Identity (Proto WaitForTxResponse) ByteString
-> ByteString -> Proto WaitForTxResponse -> Proto WaitForTxResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ TxId -> ByteString
forall a. SerialiseAsRawBytes a => a -> ByteString
serialiseToRawBytes TxId
ref
        Proto WaitForTxResponse
-> (Proto WaitForTxResponse -> Proto WaitForTxResponse)
-> Proto WaitForTxResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto WaitForTxResponse) (Proto Stage)
forall (f :: * -> *) s a.
(Functor f, HasField s "stage" a) =>
LensLike' f s a
U5c.stage LensLike' Identity (Proto WaitForTxResponse) (Proto Stage)
-> Proto Stage
-> Proto WaitForTxResponse
-> Proto WaitForTxResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Stage -> Proto Stage
forall msg. msg -> Proto msg
Proto Stage
U5c.STAGE_CONFIRMED
    | TxId
ref <- Set TxId -> [TxId]
forall a. Set a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList Set TxId
confirmed
    ]
  )
 where
  confirmed :: Set TxId
  confirmed :: Set TxId
confirmed = Set TxId
pendingRefs Set TxId -> Set TxId -> Set TxId
forall a. Ord a => Set a -> Set a -> Set a
`Set.intersection` Set TxId
block

-- | Record the pending refs now in the mempool: @STAGE_MEMPOOL@ for each
-- one that was not already announced.
announceMempool
  :: WaitForTxState
  -> Set TxId
  -> (WaitForTxState, [Proto U5c.WaitForTxResponse])
announceMempool :: WaitForTxState
-> Set TxId -> (WaitForTxState, [Proto WaitForTxResponse])
announceMempool state :: WaitForTxState
state@WaitForTxState{Set TxId
pendingRefs :: WaitForTxState -> Set TxId
pendingRefs :: Set TxId
pendingRefs, Set TxId
mempoolNotifiedRefs :: WaitForTxState -> Set TxId
mempoolNotifiedRefs :: Set TxId
mempoolNotifiedRefs} Set TxId
inMempool =
  ( WaitForTxState
state{mempoolNotifiedRefs = present}
  , [ Proto WaitForTxResponse
forall msg. Message msg => msg
defMessage
        Proto WaitForTxResponse
-> (Proto WaitForTxResponse -> Proto WaitForTxResponse)
-> Proto WaitForTxResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto WaitForTxResponse) ByteString
forall (f :: * -> *) s a.
(Functor f, HasField s "ref" a) =>
LensLike' f s a
U5c.ref LensLike' Identity (Proto WaitForTxResponse) ByteString
-> ByteString -> Proto WaitForTxResponse -> Proto WaitForTxResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ TxId -> ByteString
forall a. SerialiseAsRawBytes a => a -> ByteString
serialiseToRawBytes TxId
ref
        Proto WaitForTxResponse
-> (Proto WaitForTxResponse -> Proto WaitForTxResponse)
-> Proto WaitForTxResponse
forall a b. a -> (a -> b) -> b
& LensLike' Identity (Proto WaitForTxResponse) (Proto Stage)
forall (f :: * -> *) s a.
(Functor f, HasField s "stage" a) =>
LensLike' f s a
U5c.stage LensLike' Identity (Proto WaitForTxResponse) (Proto Stage)
-> Proto Stage
-> Proto WaitForTxResponse
-> Proto WaitForTxResponse
forall s t a b. ASetter s t a b -> b -> s -> t
.~ Stage -> Proto Stage
forall msg. msg -> Proto msg
Proto Stage
U5c.STAGE_MEMPOOL
    | TxId
ref <- Set TxId -> [TxId]
forall a. Set a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList (Set TxId -> [TxId]) -> Set TxId -> [TxId]
forall a b. (a -> b) -> a -> b
$ Set TxId
present Set TxId -> Set TxId -> Set TxId
forall a. Ord a => Set a -> Set a -> Set a
`Set.difference` Set TxId
mempoolNotifiedRefs
    ]
  )
 where
  present :: Set TxId
  present :: Set TxId
present = Set TxId
pendingRefs Set TxId -> Set TxId -> Set TxId
forall a. Ord a => Set a -> Set a -> Set a
`Set.intersection` Set TxId
inMempool

-- | The state, or 'Nothing' once no ref is pending.
whilePending :: WaitForTxState -> Maybe WaitForTxState
whilePending :: WaitForTxState -> Maybe WaitForTxState
whilePending state :: WaitForTxState
state@WaitForTxState{Set TxId
pendingRefs :: WaitForTxState -> Set TxId
pendingRefs :: Set TxId
pendingRefs}
  | Set TxId -> Bool
forall a. Set a -> Bool
Set.null Set TxId
pendingRefs = Maybe WaitForTxState
forall a. Maybe a
Nothing
  | Bool
otherwise = WaitForTxState -> Maybe WaitForTxState
forall a. a -> Maybe a
Just WaitForTxState
state