{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StandaloneKindSignatures #-}

-- | Operations that update the mempool. They are internally divided in the pure
-- and impure sides of the operation.
module Ouroboros.Consensus.Mempool.Update
  ( WhichAddTx (..)
  , implAddTx
  , implRemoveTxsEvenIfValid
  , implSyncWithLedger
  ) where

import Control.Monad (unless)
import Control.Monad.Class.MonadTimer.SI (MonadTimer, timeout)
import Control.Monad.Except (runExcept)
import Control.Tracer
import Data.Functor.Identity (Identity (Identity))
import Data.Kind (Type)
import qualified Data.List.NonEmpty as NE
import Data.Maybe (fromMaybe)
import qualified Data.Measure as Measure
import qualified Data.Set as Set
import qualified Data.Text as T
import Ouroboros.Consensus.HeaderValidation
import Ouroboros.Consensus.Ledger.Abstract
import Ouroboros.Consensus.Ledger.SupportsMempool
import Ouroboros.Consensus.Ledger.Tables.Utils
import Ouroboros.Consensus.Mempool.API
import Ouroboros.Consensus.Mempool.Capacity
import Ouroboros.Consensus.Mempool.Impl.Common
import Ouroboros.Consensus.Mempool.TxSeq (TxTicket (..))
import qualified Ouroboros.Consensus.Mempool.TxSeq as TxSeq
import Ouroboros.Consensus.Storage.LedgerDB.Forker hiding (trace)
import Ouroboros.Consensus.Util.Enclose
import Ouroboros.Consensus.Util.IOLike hiding (withMVar)
import Ouroboros.Consensus.Util.NormalForm.StrictMVar
import Ouroboros.Consensus.Util.STM
import Ouroboros.Network.Block

{-------------------------------------------------------------------------------
  Add transactions
-------------------------------------------------------------------------------}

-- | A GADT that enables the shared implementation of 'addTx' and 'testTryAddTx'.
type WhichAddTx :: (Type -> Type) -> Type
data WhichAddTx f where
  ProductionAddTx :: WhichAddTx Identity
  -- | The argument unique to 'testTryAddTx'.
  --
  -- The 'Nothing' result means the tx would not fit in the current mempool;
  -- the testing implementation gives up instead of retrying indefinitely.
  TestingAddTx :: !DiffTime -> WhichAddTx Maybe

-- | Add a single transaction to the mempool.
--
-- If there is no space, then the 'ProductionAddTx' caller will block until
-- there space, and try again, repeatedly until it succeeds. It only releases
-- the lock when this loop terminates.
--
-- If there is no space, the 'TestingAddTx' caller will immediately return
-- 'Nothing'.
implAddTx ::
  ( IOLike m
  , MonadTimer m
  , LedgerSupportsMempool blk
  , HasTxId (GenTx blk)
  ) =>
  MempoolEnv m blk ->
  WhichAddTx f ->
  -- | Whether we're acting on behalf of a remote peer or a local client.
  AddTxOnBehalfOf ->
  -- | The transaction to add to the mempool.
  GenTx blk ->
  m (f (MempoolAddTxResult blk))
implAddTx :: forall (m :: * -> *) blk (f :: * -> *).
(IOLike m, MonadTimer m, LedgerSupportsMempool blk,
 HasTxId (GenTx blk)) =>
MempoolEnv m blk
-> WhichAddTx f
-> AddTxOnBehalfOf
-> GenTx blk
-> m (f (MempoolAddTxResult blk))
implAddTx MempoolEnv m blk
mpEnv WhichAddTx f
caller AddTxOnBehalfOf
onbehalf GenTx blk
tx =
  -- To ensure fair behaviour between threads that are trying to add
  -- transactions, we make them all queue in a fifo. Only the one at the head
  -- of the queue gets to actually wait for space to get freed up in the
  -- mempool. This avoids small transactions repeatedly squeezing in ahead of
  -- larger transactions.
  --
  -- The fifo behaviour is implemented using a simple MVar. And take this
  -- MVar lock on a transaction by transaction basis. So if several threads
  -- are each trying to add several transactions, then they'll interleave at
  -- transaction granularity, not batches of transactions.
  --
  -- To add back in a bit of deliberate unfairness, we want to prioritise
  -- transactions being added on behalf of local clients, over ones being
  -- added on behalf of remote peers. We do this by using a pair of mvar
  -- fifos: remote peers must wait on both mvars, while local clients only
  -- need to wait on the second.
  case AddTxOnBehalfOf
onbehalf of
    AddTxOnBehalfOf
AddTxForRemotePeer ->
      StrictMVar m ()
-> (() -> m (f (MempoolAddTxResult blk)))
-> m (f (MempoolAddTxResult blk))
forall (m :: * -> *) a b.
MonadMVar m =>
StrictMVar m a -> (a -> m b) -> m b
withMVar StrictMVar m ()
remoteFifo ((() -> m (f (MempoolAddTxResult blk)))
 -> m (f (MempoolAddTxResult blk)))
-> (() -> m (f (MempoolAddTxResult blk)))
-> m (f (MempoolAddTxResult blk))
forall a b. (a -> b) -> a -> b
$ \() ->
        StrictMVar m ()
-> (() -> m (f (MempoolAddTxResult blk)))
-> m (f (MempoolAddTxResult blk))
forall (m :: * -> *) a b.
MonadMVar m =>
StrictMVar m a -> (a -> m b) -> m b
withMVar StrictMVar m ()
allFifo ((() -> m (f (MempoolAddTxResult blk)))
 -> m (f (MempoolAddTxResult blk)))
-> (() -> m (f (MempoolAddTxResult blk)))
-> m (f (MempoolAddTxResult blk))
forall a b. (a -> b) -> a -> b
$ \() ->
          -- This action can also block. Holding the MVars means
          -- there is only a single such thread blocking at once.
          m (f (MempoolAddTxResult blk))
implAddTx'
    AddTxOnBehalfOf
AddTxForLocalClient ->
      StrictMVar m ()
-> (() -> m (f (MempoolAddTxResult blk)))
-> m (f (MempoolAddTxResult blk))
forall (m :: * -> *) a b.
MonadMVar m =>
StrictMVar m a -> (a -> m b) -> m b
withMVar StrictMVar m ()
allFifo ((() -> m (f (MempoolAddTxResult blk)))
 -> m (f (MempoolAddTxResult blk)))
-> (() -> m (f (MempoolAddTxResult blk)))
-> m (f (MempoolAddTxResult blk))
forall a b. (a -> b) -> a -> b
$ \() ->
        -- As above but skip the first MVar fifo so we will get
        -- service sooner if there's lots of other remote
        -- threads waiting.
        m (f (MempoolAddTxResult blk))
implAddTx'
 where
  MempoolEnv
    { mpEnvAddTxsRemoteFifo :: forall (m :: * -> *) blk. MempoolEnv m blk -> StrictMVar m ()
mpEnvAddTxsRemoteFifo = StrictMVar m ()
remoteFifo
    , mpEnvAddTxsAllFifo :: forall (m :: * -> *) blk. MempoolEnv m blk -> StrictMVar m ()
mpEnvAddTxsAllFifo = StrictMVar m ()
allFifo
    , mpEnvTracer :: forall (m :: * -> *) blk.
MempoolEnv m blk -> Tracer m (TraceEventMempool blk)
mpEnvTracer = Tracer m (TraceEventMempool blk)
trcr
    } = MempoolEnv m blk
mpEnv

  implAddTx' :: m (f (MempoolAddTxResult blk))
implAddTx' = do
    x <- MempoolEnv m blk
-> WhichAddTx f
-> WhetherToIntervene
-> GenTx blk
-> m (f (TransactionProcessed blk))
forall (m :: * -> *) blk (f :: * -> *).
(LedgerSupportsMempool blk, HasTxId (GenTx blk), IOLike m,
 MonadTimer m) =>
MempoolEnv m blk
-> WhichAddTx f
-> WhetherToIntervene
-> GenTx blk
-> m (f (TransactionProcessed blk))
doAddTx MempoolEnv m blk
mpEnv WhichAddTx f
caller WhetherToIntervene
wti GenTx blk
tx
    case (caller, x) of
      (WhichAddTx f
ProductionAddTx, Identity (TransactionProcessingResult Maybe (InternalState blk)
_ MempoolAddTxResult blk
result TraceEventMempool blk
ev)) -> do
        Tracer m (TraceEventMempool blk) -> TraceEventMempool blk -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceEventMempool blk)
trcr TraceEventMempool blk
ev
        f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk))
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return (f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk)))
-> f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk))
forall a b. (a -> b) -> a -> b
$ MempoolAddTxResult blk -> Identity (MempoolAddTxResult blk)
forall a. a -> Identity a
Identity MempoolAddTxResult blk
result
      (TestingAddTx DiffTime
_, Just (TransactionProcessingResult Maybe (InternalState blk)
_ MempoolAddTxResult blk
result TraceEventMempool blk
ev)) -> do
        Tracer m (TraceEventMempool blk) -> TraceEventMempool blk -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceEventMempool blk)
trcr TraceEventMempool blk
ev
        f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk))
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return (f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk)))
-> f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk))
forall a b. (a -> b) -> a -> b
$ MempoolAddTxResult blk -> Maybe (MempoolAddTxResult blk)
forall a. a -> Maybe a
Just MempoolAddTxResult blk
result
      (TestingAddTx DiffTime
_, f (TransactionProcessed blk)
Maybe (TransactionProcessed blk)
Nothing) -> f (MempoolAddTxResult blk) -> m (f (MempoolAddTxResult blk))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure f (MempoolAddTxResult blk)
Maybe (MempoolAddTxResult blk)
forall a. Maybe a
Nothing

  wti :: WhetherToIntervene
  wti :: WhetherToIntervene
wti = case AddTxOnBehalfOf
onbehalf of
    AddTxOnBehalfOf
AddTxForRemotePeer -> WhetherToIntervene
DoNotIntervene
    AddTxOnBehalfOf
AddTxForLocalClient -> WhetherToIntervene
Intervene

-- | Tried to add a transaction, was it processed or is there no space left?
data TriedToAddTx blk
  = -- | Adding the next transaction would put the mempool over capacity.
    NotEnoughSpaceLeft
  | -- | The tx was rejected based on the result 'txMeasure'; we didn't even
    -- try to validate the tx.
    NotProcessed (TransactionProcessed blk)
  | -- | Implementation detail: this argument is strict in order to prevent
    -- this constructor from being floated out of both branches of the case
    -- in 'pureTryAddTx', since that function is the argument of a 'timeout'
    -- call in 'doAddTx'.
    Processed !(DiffTimeMeasure -> TransactionProcessed blk)

-- | The new state, if the transaction was accepted
data TransactionProcessed blk
  = TransactionProcessingResult
      -- | If the transaction was accepted, the new state that can be written to
      -- the TVar.
      (Maybe (InternalState blk))
      -- | The result of trying to add the transaction to the mempool.
      (MempoolAddTxResult blk)
      -- | The event emitted by the operation.
      (TraceEventMempool blk)

-- | This function returns whether the transaction was added or rejected, and
-- will block if the mempool is full.
--
-- This function returns whether the transaction was added or rejected, or if
-- the Mempool capacity is reached. See 'implAddTx' for a function that blocks
-- in case the Mempool capacity is reached.
--
-- Transactions are added one by one, updating the Mempool each time one was
-- added successfully.
--
-- See the necessary invariants on the Haddock for 'API.addTxs'.
--
-- INVARIANT: The code needs that read and writes on the state are coupled
-- together or inconsistencies will arise.
doAddTx ::
  forall m blk f.
  ( LedgerSupportsMempool blk
  , HasTxId (GenTx blk)
  , IOLike m
  , MonadTimer m
  ) =>
  MempoolEnv m blk ->
  WhichAddTx f ->
  WhetherToIntervene ->
  -- | The transaction to add to the mempool.
  GenTx blk ->
  m (f (TransactionProcessed blk))
doAddTx :: forall (m :: * -> *) blk (f :: * -> *).
(LedgerSupportsMempool blk, HasTxId (GenTx blk), IOLike m,
 MonadTimer m) =>
MempoolEnv m blk
-> WhichAddTx f
-> WhetherToIntervene
-> GenTx blk
-> m (f (TransactionProcessed blk))
doAddTx MempoolEnv m blk
mpEnv WhichAddTx f
caller WhetherToIntervene
wti GenTx blk
tx = do
  Maybe MempoolSize -> m (f (TransactionProcessed blk))
doAddTx' Maybe MempoolSize
forall a. Maybe a
Nothing
 where
  MempoolEnv
    { mpEnvForker :: forall (m :: * -> *) blk.
MempoolEnv m blk -> StrictMVar m (ReadOnlyForker m LedgerState blk)
mpEnvForker = StrictMVar m (ReadOnlyForker m LedgerState blk)
forker
    , mpEnvLedgerCfg :: forall (m :: * -> *) blk. MempoolEnv m blk -> LedgerConfig blk
mpEnvLedgerCfg = LedgerConfig blk
cfg
    , mpEnvStateVar :: forall (m :: * -> *) blk.
MempoolEnv m blk -> StrictTMVar m (InternalState blk)
mpEnvStateVar = StrictTMVar m (InternalState blk)
istate
    , mpEnvTracer :: forall (m :: * -> *) blk.
MempoolEnv m blk -> Tracer m (TraceEventMempool blk)
mpEnvTracer = Tracer m (TraceEventMempool blk)
trcr
    , mpEnvTimeoutConfig :: forall (m :: * -> *) blk.
MempoolEnv m blk -> Maybe MempoolTimeoutConfig
mpEnvTimeoutConfig = Maybe MempoolTimeoutConfig
mbToCfg
    } = MempoolEnv m blk
mpEnv

  doAddTx' :: Maybe MempoolSize -> m (f (TransactionProcessed blk))
  doAddTx' :: Maybe MempoolSize -> m (f (TransactionProcessed blk))
doAddTx' Maybe MempoolSize
mbPrevSize = do
    Tracer m (TraceEventMempool blk) -> TraceEventMempool blk -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceEventMempool blk)
trcr (TraceEventMempool blk -> m ()) -> TraceEventMempool blk -> m ()
forall a b. (a -> b) -> a -> b
$ GenTx blk -> TraceEventMempool blk
forall blk. GenTx blk -> TraceEventMempool blk
TraceMempoolAttemptingAdd GenTx blk
tx

    -- If retrying, wait until the mempool size changes before attempting to
    -- add the tx again
    let additionalCheck :: InternalState blk -> STM m ()
additionalCheck InternalState blk
is =
          case Maybe MempoolSize
mbPrevSize of
            Maybe MempoolSize
Nothing -> () -> STM m ()
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            Just MempoolSize
prevSize -> Bool -> STM m ()
forall (m :: * -> *). MonadSTM m => Bool -> STM m ()
check (Bool -> STM m ()) -> Bool -> STM m ()
forall a b. (a -> b) -> a -> b
$ InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is MempoolSize -> MempoolSize -> Bool
forall a. Eq a => a -> a -> Bool
/= MempoolSize
prevSize

    eRes <- StrictTMVar m (InternalState blk)
-> (InternalState blk -> STM m ())
-> (InternalState blk
    -> ()
    -> m (Either MempoolSize (TransactionProcessed blk),
          InternalState blk))
-> m (Either MempoolSize (TransactionProcessed blk))
forall (m :: * -> *) a b c.
IOLike m =>
StrictTMVar m a -> (a -> STM m b) -> (a -> b -> m (c, a)) -> m c
withTMVarAnd StrictTMVar m (InternalState blk)
istate InternalState blk -> STM m ()
additionalCheck ((InternalState blk
  -> ()
  -> m (Either MempoolSize (TransactionProcessed blk),
        InternalState blk))
 -> m (Either MempoolSize (TransactionProcessed blk)))
-> (InternalState blk
    -> ()
    -> m (Either MempoolSize (TransactionProcessed blk),
          InternalState blk))
-> m (Either MempoolSize (TransactionProcessed blk))
forall a b. (a -> b) -> a -> b
$
      \InternalState blk
is () ->
        case Except (ApplyTxErr blk) (TxMeasurePhase1 blk)
-> Either (ApplyTxErr blk) (TxMeasurePhase1 blk)
forall e a. Except e a -> Either e a
runExcept (Except (ApplyTxErr blk) (TxMeasurePhase1 blk)
 -> Either (ApplyTxErr blk) (TxMeasurePhase1 blk))
-> Except (ApplyTxErr blk) (TxMeasurePhase1 blk)
-> Either (ApplyTxErr blk) (TxMeasurePhase1 blk)
forall a b. (a -> b) -> a -> b
$ LedgerConfig blk
-> TickedLedgerState blk EmptyMK
-> GenTx blk
-> Except (ApplyTxErr blk) (TxMeasurePhase1 blk)
forall blk.
TxLimits blk =>
LedgerConfig blk
-> TickedLedgerState blk EmptyMK
-> GenTx blk
-> Except (ApplyTxErr blk) (TxMeasurePhase1 blk)
txMeasurePhase1 LedgerConfig blk
cfg (Ticked LedgerState blk DiffMK -> TickedLedgerState blk EmptyMK
forall (l :: * -> MapKind -> *) blk (mk :: MapKind).
HasLedgerTables l blk =>
l blk mk -> l blk EmptyMK
forgetLedgerTables (Ticked LedgerState blk DiffMK -> TickedLedgerState blk EmptyMK)
-> Ticked LedgerState blk DiffMK -> TickedLedgerState blk EmptyMK
forall a b. (a -> b) -> a -> b
$ InternalState blk -> Ticked LedgerState blk DiffMK
forall blk. InternalState blk -> TickedLedgerState blk DiffMK
isLedgerState InternalState blk
is) GenTx blk
tx of
          Left ApplyTxErr blk
err ->
            -- The transaction does not have a valid measure (eg its ExUnits is
            -- greater than what this ledger state allows for a single transaction).
            (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
              ( TransactionProcessed blk
-> Either MempoolSize (TransactionProcessed blk)
forall a b. b -> Either a b
Right
                  ( Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
forall blk.
Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
TransactionProcessingResult
                      Maybe (InternalState blk)
forall a. Maybe a
Nothing
                      (GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
forall blk. GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
MempoolTxRejected GenTx blk
tx ApplyTxErr blk
err)
                      ( GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
forall blk.
GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
TraceMempoolRejectedTx
                          GenTx blk
tx
                          ApplyTxErr blk
err
                          MempoolRejectionDetails
MempoolRejectedByLedger
                          (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is)
                      )
                  )
              , InternalState blk
is
              )
          Right TxMeasurePhase1 blk
txsz1
            | let currentSize :: TxMeasureWithDiffTime blk
currentSize = TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> TxMeasureWithDiffTime blk
forall sz tx. Measure sz => TxSeq sz tx -> sz
TxSeq.toSize (InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is)
            , Bool -> Bool
not (Bool -> Bool) -> Bool -> Bool
forall a b. (a -> b) -> a -> b
$
                TxMeasureWithDiffTime blk
currentSize
                  TxMeasureWithDiffTime blk -> TxMeasureWithDiffTime blk -> Bool
forall a. Measure a => a -> a -> Bool
Measure.<= TxMeasureWithDiffTime blk
currentSize
                  TxMeasureWithDiffTime blk
-> TxMeasureWithDiffTime blk -> TxMeasureWithDiffTime blk
forall a. Measure a => a -> a -> a
`Measure.plus` TxMeasure blk -> DiffTimeMeasure -> TxMeasureWithDiffTime blk
forall blk.
TxMeasure blk -> DiffTimeMeasure -> TxMeasureWithDiffTime blk
MkTxMeasureWithDiffTime (TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
forall blk.
TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
TxMeasure TxMeasurePhase1 blk
txsz1 TxMeasurePhase2 blk
forall a. Measure a => a
Measure.zero) DiffTimeMeasure
forall a. Measure a => a
Measure.zero ->
                (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (MempoolSize -> Either MempoolSize (TransactionProcessed blk)
forall a b. a -> Either a b
Left (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is), InternalState blk
is)
            | let currentSize :: TxMeasureWithDiffTime blk
currentSize = TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> TxMeasureWithDiffTime blk
forall sz tx. Measure sz => TxSeq sz tx -> sz
TxSeq.toSize (InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is)
            , let MkTxMeasureWithDiffTime TxMeasure blk
txssz DiffTimeMeasure
_txsdifftime = TxMeasureWithDiffTime blk
currentSize
            , Bool -> Bool
not (Bool -> Bool) -> Bool -> Bool
forall a b. (a -> b) -> a -> b
$ TxMeasure blk
txssz TxMeasure blk -> TxMeasure blk -> TxMeasure blk
forall a. Measure a => a -> a -> a
`Measure.plus` (TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
forall blk.
TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
TxMeasure TxMeasurePhase1 blk
txsz1 TxMeasurePhase2 blk
forall a. Measure a => a
Measure.zero) TxMeasure blk -> TxMeasure blk -> Bool
forall a. Measure a => a -> a -> Bool
Measure.<= InternalState blk -> TxMeasure blk
forall blk. InternalState blk -> TxMeasure blk
isCapacity InternalState blk
is ->
                (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (MempoolSize -> Either MempoolSize (TransactionProcessed blk)
forall a b. a -> Either a b
Left (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is), InternalState blk
is)
            | Bool
otherwise -> do
                frkr <- StrictMVar m (ReadOnlyForker m LedgerState blk)
-> m (ReadOnlyForker m LedgerState blk)
forall (m :: * -> *) a. MonadMVar m => StrictMVar m a -> m a
readMVar StrictMVar m (ReadOnlyForker m LedgerState blk)
forker
                tbs <- roforkerReadTables frkr (getTransactionKeySets tx)
                before <- getMonotonicTime
                mbX <- do
                  let f m (TriedToAddTx blk)
m = case Maybe MempoolTimeoutConfig
mbToCfg of
                        Maybe MempoolTimeoutConfig
Nothing -> TriedToAddTx blk -> Maybe (TriedToAddTx blk)
forall a. a -> Maybe a
Just (TriedToAddTx blk -> Maybe (TriedToAddTx blk))
-> m (TriedToAddTx blk) -> m (Maybe (TriedToAddTx blk))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m (TriedToAddTx blk)
m
                        Just MempoolTimeoutConfig
toCfg -> DiffTime -> m (TriedToAddTx blk) -> m (Maybe (TriedToAddTx blk))
forall a. DiffTime -> m a -> m (Maybe a)
forall (m :: * -> *) a.
MonadTimer m =>
DiffTime -> m a -> m (Maybe a)
timeout (MempoolTimeoutConfig -> DiffTime
mempoolTimeoutHard MempoolTimeoutConfig
toCfg) m (TriedToAddTx blk)
m
                  f $ do
                    x <- evaluate $ pureTryAddTx mpEnv cfg wti tx is tbs txsz1
                    case (caller, x) of
                      (TestingAddTx DiffTime
testDiffTime, Processed{}) -> do
                        after <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
                        let sofar = Time
after Time -> Time -> DiffTime
`diffTime` Time
before
                        threadDelay $ testDiffTime - min testDiffTime sofar
                      -- Note that @sofar == 0@ always and this 'threadDelay' would
                      -- be perfectly precise in the @IOSim@ monad. Unfortunately,
                      -- the state machines tests are still only in IO.
                      (WhichAddTx f, TriedToAddTx blk)
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
                    pure x
                dur <- do
                  -- Note that both the hard 'timeout' and the soft duration check use
                  -- the actual monotonic clock measurements instead of simply
                  -- deferring to 'TestingAddTx'. This means the test will fail if the
                  -- 'timeout' and the monotonic clock measurement primitives are not
                  -- as precise as the test expects (recall that the test suite chooses
                  -- intended validation times that are not "too close" to the
                  -- thresholds).
                  after <- getMonotonicTime
                  pure $ after `diffTime` before
                let rejectBecauseOfTimeoutSoft ApplyTxErr blk
txerr = do
                      let outcome :: TransactionProcessed blk
outcome =
                            Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
forall blk.
Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
TransactionProcessingResult
                              Maybe (InternalState blk)
forall a. Maybe a
Nothing
                              (GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
forall blk. GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
MempoolTxRejected GenTx blk
tx ApplyTxErr blk
txerr)
                              (TraceEventMempool blk -> TransactionProcessed blk)
-> TraceEventMempool blk -> TransactionProcessed blk
forall a b. (a -> b) -> a -> b
$ GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
forall blk.
GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
TraceMempoolRejectedTx
                                GenTx blk
tx
                                ApplyTxErr blk
txerr
                                (DiffTime -> MempoolRejectionDetails
MempoolRejectedByTimeoutSoft DiffTime
dur)
                                (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is)
                      (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TransactionProcessed blk
-> Either MempoolSize (TransactionProcessed blk)
forall a b. b -> Either a b
Right TransactionProcessed blk
outcome, InternalState blk
is)
                    mbTimeoutSoftTxErr =
                      -- This @txerr@ is not available in historical Cardano eras, but
                      -- it is starting from Conway. So this rejection will be disabled
                      -- prior to Conway. Which is irrelevant, since mainnet is already
                      -- in Conway.
                      let txt :: Text
txt = String -> Text
T.pack (String -> Text) -> String -> Text
forall a b. (a -> b) -> a -> b
$ String
"MempoolTxTooSlow (" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> DiffTime -> String
forall a. Show a => a -> String
show DiffTime
dur String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
") " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> TxId (GenTx blk) -> String
forall a. Show a => a -> String
show (GenTx blk -> TxId (GenTx blk)
forall tx. HasTxId tx => tx -> TxId tx
txId GenTx blk
tx)
                       in Ticked LedgerState blk DiffMK -> Text -> Maybe (ApplyTxErr blk)
forall blk (mk :: MapKind).
LedgerSupportsMempool blk =>
TickedLedgerState blk mk -> Text -> Maybe (ApplyTxErr blk)
forall (mk :: MapKind).
TickedLedgerState blk mk -> Text -> Maybe (ApplyTxErr blk)
mkMempoolApplyTxError (InternalState blk -> Ticked LedgerState blk DiffMK
forall blk. InternalState blk -> TickedLedgerState blk DiffMK
isLedgerState InternalState blk
is) Text
txt
                case mbX of
                  Maybe (TriedToAddTx blk)
Nothing -> case (WhetherToIntervene
wti, Maybe (ApplyTxErr blk)
mbTimeoutSoftTxErr) of
                    (WhetherToIntervene
Intervene, Just ApplyTxErr blk
txerr) -> do
                      ApplyTxErr blk
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
rejectBecauseOfTimeoutSoft ApplyTxErr blk
txerr
                    (WhetherToIntervene, Maybe (ApplyTxErr blk))
_ -> do
                      -- Either they're not a local client or the era doesn't allow for
                      -- soft rejections.
                      ExnMempoolTimeout
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (ExnMempoolTimeout
 -> m (Either MempoolSize (TransactionProcessed blk),
       InternalState blk))
-> ExnMempoolTimeout
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a b. (a -> b) -> a -> b
$ DiffTime -> GenTx blk -> ExnMempoolTimeout
forall blk.
Show (GenTx blk) =>
DiffTime -> GenTx blk -> ExnMempoolTimeout
MkExnMempoolTimeout DiffTime
dur GenTx blk
tx
                  Just TriedToAddTx blk
_
                    | Just MempoolTimeoutConfig
toCfg <- Maybe MempoolTimeoutConfig
mbToCfg
                    , DiffTime
dur DiffTime -> DiffTime -> Bool
forall a. Ord a => a -> a -> Bool
> MempoolTimeoutConfig -> DiffTime
mempoolTimeoutSoft MempoolTimeoutConfig
toCfg
                    , Just ApplyTxErr blk
txerr <- Maybe (ApplyTxErr blk)
mbTimeoutSoftTxErr -> do
                        ApplyTxErr blk
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
rejectBecauseOfTimeoutSoft ApplyTxErr blk
txerr
                  Just TriedToAddTx blk
NotEnoughSpaceLeft -> do
                    (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (MempoolSize -> Either MempoolSize (TransactionProcessed blk)
forall a b. a -> Either a b
Left (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is), InternalState blk
is)
                  Just (NotProcessed TransactionProcessed blk
outcome) -> do
                    let TransactionProcessingResult Maybe (InternalState blk)
is' MempoolAddTxResult blk
_ TraceEventMempool blk
_ = TransactionProcessed blk
outcome
                    (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TransactionProcessed blk
-> Either MempoolSize (TransactionProcessed blk)
forall a b. b -> Either a b
Right TransactionProcessed blk
outcome, InternalState blk -> Maybe (InternalState blk) -> InternalState blk
forall a. a -> Maybe a -> a
fromMaybe InternalState blk
is Maybe (InternalState blk)
is')
                  Just (Processed DiffTimeMeasure -> TransactionProcessed blk
mkResult) -> do
                    let outcome :: TransactionProcessed blk
outcome = DiffTimeMeasure -> TransactionProcessed blk
mkResult (DiffTimeMeasure -> TransactionProcessed blk)
-> DiffTimeMeasure -> TransactionProcessed blk
forall a b. (a -> b) -> a -> b
$ DiffTime -> DiffTimeMeasure
FiniteDiffTimeMeasure (DiffTime -> DiffTimeMeasure) -> DiffTime -> DiffTimeMeasure
forall a b. (a -> b) -> a -> b
$ case WhichAddTx f
caller of
                          WhichAddTx f
ProductionAddTx -> DiffTime
dur
                          TestingAddTx DiffTime
testDiffTime ->
                            -- For the sake of an accurate cumulative measure, pretend
                            -- the tx took exactly as long to validate as the test
                            -- suite intended.
                            --
                            -- Note that @testDiffTime == dur@ always in @IOSim@.
                            -- Unfortunately, the state machines tests are still only
                            -- in IO.
                            DiffTime
testDiffTime
                        TransactionProcessingResult Maybe (InternalState blk)
is' MempoolAddTxResult blk
_ TraceEventMempool blk
_ = TransactionProcessed blk
outcome
                    (Either MempoolSize (TransactionProcessed blk), InternalState blk)
-> m (Either MempoolSize (TransactionProcessed blk),
      InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TransactionProcessed blk
-> Either MempoolSize (TransactionProcessed blk)
forall a b. b -> Either a b
Right TransactionProcessed blk
outcome, InternalState blk -> Maybe (InternalState blk) -> InternalState blk
forall a. a -> Maybe a -> a
fromMaybe InternalState blk
is Maybe (InternalState blk)
is')
    case (caller, eRes) of
      (WhichAddTx f
ProductionAddTx, Either MempoolSize (TransactionProcessed blk)
_) -> (MempoolSize -> m (f (TransactionProcessed blk)))
-> (TransactionProcessed blk -> m (f (TransactionProcessed blk)))
-> Either MempoolSize (TransactionProcessed blk)
-> m (f (TransactionProcessed blk))
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Maybe MempoolSize -> m (f (TransactionProcessed blk))
doAddTx' (Maybe MempoolSize -> m (f (TransactionProcessed blk)))
-> (MempoolSize -> Maybe MempoolSize)
-> MempoolSize
-> m (f (TransactionProcessed blk))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. MempoolSize -> Maybe MempoolSize
forall a. a -> Maybe a
Just) (f (TransactionProcessed blk) -> m (f (TransactionProcessed blk))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (f (TransactionProcessed blk) -> m (f (TransactionProcessed blk)))
-> (TransactionProcessed blk -> f (TransactionProcessed blk))
-> TransactionProcessed blk
-> m (f (TransactionProcessed blk))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TransactionProcessed blk -> f (TransactionProcessed blk)
TransactionProcessed blk -> Identity (TransactionProcessed blk)
forall a. a -> Identity a
Identity) Either MempoolSize (TransactionProcessed blk)
eRes
      (TestingAddTx DiffTime
_, Left MempoolSize
_) -> f (TransactionProcessed blk) -> m (f (TransactionProcessed blk))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure f (TransactionProcessed blk)
Maybe (TransactionProcessed blk)
forall a. Maybe a
Nothing
      (TestingAddTx DiffTime
_, Right TransactionProcessed blk
x) -> f (TransactionProcessed blk) -> m (f (TransactionProcessed blk))
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (f (TransactionProcessed blk) -> m (f (TransactionProcessed blk)))
-> f (TransactionProcessed blk) -> m (f (TransactionProcessed blk))
forall a b. (a -> b) -> a -> b
$ TransactionProcessed blk -> Maybe (TransactionProcessed blk)
forall a. a -> Maybe a
Just TransactionProcessed blk
x

pureTryAddTx ::
  ( LedgerSupportsMempool blk
  , HasTxId (GenTx blk)
  ) =>
  MempoolEnv m blk ->
  -- | The ledger configuration.
  LedgerCfg LedgerState blk ->
  WhetherToIntervene ->
  -- | The transaction to add to the mempool.
  GenTx blk ->
  -- | The current internal state of the mempool.
  InternalState blk ->
  LedgerTables blk ValuesMK ->
  TxMeasurePhase1 blk ->
  TriedToAddTx blk
pureTryAddTx :: forall blk (m :: * -> *).
(LedgerSupportsMempool blk, HasTxId (GenTx blk)) =>
MempoolEnv m blk
-> LedgerCfg LedgerState blk
-> WhetherToIntervene
-> GenTx blk
-> InternalState blk
-> LedgerTables blk ValuesMK
-> TxMeasurePhase1 blk
-> TriedToAddTx blk
pureTryAddTx MempoolEnv m blk
mpEnv LedgerCfg LedgerState blk
cfg WhetherToIntervene
wti GenTx blk
tx InternalState blk
is LedgerTables blk ValuesMK
values TxMeasurePhase1 blk
p1TxMeasure =
  let MempoolEnv
        { mpEnvTimeoutConfig :: forall (m :: * -> *) blk.
MempoolEnv m blk -> Maybe MempoolTimeoutConfig
mpEnvTimeoutConfig = Maybe MempoolTimeoutConfig
mbToCfg
        } = MempoolEnv m blk
mpEnv

      st :: TickedLedgerState blk ValuesMK
st =
        LedgerTables blk ValuesMK
-> LedgerTables blk KeysMK
-> TickedLedgerState blk DiffMK
-> TickedLedgerState blk ValuesMK
forall blk.
LedgerSupportsMempool blk =>
LedgerTables blk ValuesMK
-> LedgerTables blk KeysMK
-> TickedLedgerState blk DiffMK
-> TickedLedgerState blk ValuesMK
applyMempoolDiffs
          LedgerTables blk ValuesMK
values
          (GenTx blk -> LedgerTables blk KeysMK
forall blk.
LedgerSupportsMempool blk =>
GenTx blk -> LedgerTables blk KeysMK
getTransactionKeySets GenTx blk
tx)
          (InternalState blk -> TickedLedgerState blk DiffMK
forall blk. InternalState blk -> TickedLedgerState blk DiffMK
isLedgerState InternalState blk
is)
   in case Except (ApplyTxErr blk) (TxMeasurePhase2 blk)
-> Either (ApplyTxErr blk) (TxMeasurePhase2 blk)
forall e a. Except e a -> Either e a
runExcept (Except (ApplyTxErr blk) (TxMeasurePhase2 blk)
 -> Either (ApplyTxErr blk) (TxMeasurePhase2 blk))
-> Except (ApplyTxErr blk) (TxMeasurePhase2 blk)
-> Either (ApplyTxErr blk) (TxMeasurePhase2 blk)
forall a b. (a -> b) -> a -> b
$ LedgerCfg LedgerState blk
-> TickedLedgerState blk ValuesMK
-> GenTx blk
-> Except (ApplyTxErr blk) (TxMeasurePhase2 blk)
forall blk.
TxLimits blk =>
LedgerConfig blk
-> TickedLedgerState blk ValuesMK
-> GenTx blk
-> Except (ApplyTxErr blk) (TxMeasurePhase2 blk)
txMeasurePhase2 LedgerCfg LedgerState blk
cfg TickedLedgerState blk ValuesMK
st GenTx blk
tx of
        Left ApplyTxErr blk
err ->
          -- The transaction does not have a valid measure (eg its ExUnits is
          -- greater than what this ledger state allows for a single transaction).
          --
          -- It might seem simpler to remove the failure case from 'txMeasure' and
          -- simply fully validate the tx before determining whether it'd fit in
          -- the mempool; that way we could reject invalid txs ASAP. However, for a
          -- valid tx, we'd pay that validation cost every time the node's
          -- selection changed, even if the tx wouldn't fit. So it'd very much be
          -- as if the mempool were effectively over capacity! What's worse, each
          -- attempt would not be using 'extendVRPrevApplied'.
          TransactionProcessed blk -> TriedToAddTx blk
forall blk. TransactionProcessed blk -> TriedToAddTx blk
NotProcessed (TransactionProcessed blk -> TriedToAddTx blk)
-> TransactionProcessed blk -> TriedToAddTx blk
forall a b. (a -> b) -> a -> b
$
            Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
forall blk.
Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
TransactionProcessingResult
              Maybe (InternalState blk)
forall a. Maybe a
Nothing
              (GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
forall blk. GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
MempoolTxRejected GenTx blk
tx ApplyTxErr blk
err)
              ( GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
forall blk.
GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
TraceMempoolRejectedTx
                  GenTx blk
tx
                  ApplyTxErr blk
err
                  MempoolRejectionDetails
MempoolRejectedByLedger
                  (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is)
              )
        Right TxMeasurePhase2 blk
txsz
          -- Check for overflow
          --
          -- No measure of a transaction can ever be negative, so the only way
          -- adding two measures could result in a smaller measure is if some
          -- modular arithmetic overflowed. Also, overflow necessarily yields a
          -- lesser result, since adding 'maxBound' is modularly equivalent to
          -- subtracting one. Recall that we're checking each individual addition.
          --
          -- We assume that the 'txMeasure' limit and the mempool capacity
          -- 'isCapacity' are much smaller than the modulus, and so this should
          -- never happen. Despite that, blocking until adding the transaction
          -- doesn't overflow seems like a reasonable way to handle this case.
          | Bool -> Bool
not (Bool -> Bool) -> Bool -> Bool
forall a b. (a -> b) -> a -> b
$
              TxMeasureWithDiffTime blk
currentSize
                TxMeasureWithDiffTime blk -> TxMeasureWithDiffTime blk -> Bool
forall a. Measure a => a -> a -> Bool
Measure.<= TxMeasureWithDiffTime blk
currentSize
                TxMeasureWithDiffTime blk
-> TxMeasureWithDiffTime blk -> TxMeasureWithDiffTime blk
forall a. Measure a => a -> a -> a
`Measure.plus` TxMeasure blk -> DiffTimeMeasure -> TxMeasureWithDiffTime blk
forall blk.
TxMeasure blk -> DiffTimeMeasure -> TxMeasureWithDiffTime blk
MkTxMeasureWithDiffTime (TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
forall blk.
TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
TxMeasure TxMeasurePhase1 blk
p1TxMeasure TxMeasurePhase2 blk
txsz) DiffTimeMeasure
forall a. Measure a => a
Measure.zero ->
              TriedToAddTx blk
forall blk. TriedToAddTx blk
NotEnoughSpaceLeft
          -- We add the transaction if and only if it wouldn't overrun any component
          -- of the mempool capacity.
          --
          -- In the past, this condition was instead @TxSeq.toSize (isTxs is) <
          -- isCapacity is@. Thus the effective capacity of the mempool was
          -- actually one increment less than the reported capacity plus one
          -- transaction. That subtlety's cost paid for two benefits.
          --
          -- First, the absence of addition avoids a risk of overflow, since the
          -- transaction's sizes (eg ExUnits) have not yet been bounded by
          -- validation (which presumably enforces a low enough bound that any
          -- reasonably-sized mempool would never overflow the representation's
          -- 'maxBound').
          --
          -- Second, it is more fair, since it does not depend on the transaction
          -- at all. EG a large transaction might struggle to win the race against
          -- a firehose of tiny transactions.
          --
          -- However, we prefer to avoid the subtlety. Overflow is handled by the
          -- previous guard. And fairness is already ensured elsewhere (the 'MVar's
          -- in 'implAddTx' --- which the "Test.Consensus.Mempool.Fairness" test
          -- exercises). Moreover, the notion of "is under capacity" becomes
          -- difficult to assess independently of the pending tx when the measure
          -- is multi-dimensional; both typical options (any component is not full
          -- or every component is not full) lead to some confusing behaviors
          -- (denying some txs that would "obviously" fit and accepting some txs
          -- that "obviously" don't, respectively).
          --
          -- Even with the overflow handler, it's important that 'txMeasure'
          -- returns a well-bounded result. Otherwise, if an adversarial tx arrived
          -- that could't even fit in an empty mempool, then that thread would
          -- never release the 'MVar'. In particular, we tacitly assume here that a
          -- tx that wouldn't even fit in an empty mempool would be rejected by
          -- 'txMeasure'.
          | let MkTxMeasureWithDiffTime TxMeasure blk
txssz DiffTimeMeasure
_txsdifftime = TxMeasureWithDiffTime blk
currentSize
          , Bool -> Bool
not (Bool -> Bool) -> Bool -> Bool
forall a b. (a -> b) -> a -> b
$ TxMeasure blk
txssz TxMeasure blk -> TxMeasure blk -> TxMeasure blk
forall a. Measure a => a -> a -> a
`Measure.plus` (TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
forall blk.
TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
TxMeasure TxMeasurePhase1 blk
p1TxMeasure TxMeasurePhase2 blk
txsz) TxMeasure blk -> TxMeasure blk -> Bool
forall a. Measure a => a -> a -> Bool
Measure.<= InternalState blk -> TxMeasure blk
forall blk. InternalState blk -> TxMeasure blk
isCapacity InternalState blk
is ->
              TriedToAddTx blk
forall blk. TriedToAddTx blk
NotEnoughSpaceLeft
          | Just MempoolTimeoutConfig
toCfg <- Maybe MempoolTimeoutConfig
mbToCfg
          , let MkTxMeasureWithDiffTime TxMeasure blk
_txssz DiffTimeMeasure
txsdifftime = TxMeasureWithDiffTime blk
currentSize
          , Bool -> Bool
not (Bool -> Bool) -> Bool -> Bool
forall a b. (a -> b) -> a -> b
$ DiffTimeMeasure
txsdifftime DiffTimeMeasure -> DiffTimeMeasure -> Bool
forall a. Measure a => a -> a -> Bool
Measure.<= DiffTime -> DiffTimeMeasure
FiniteDiffTimeMeasure (MempoolTimeoutConfig -> DiffTime
mempoolTimeoutCapacity MempoolTimeoutConfig
toCfg) ->
              TriedToAddTx blk
forall blk. TriedToAddTx blk
NotEnoughSpaceLeft
          | Bool
otherwise ->
              case LedgerCfg LedgerState blk
-> WhetherToIntervene
-> GenTx blk
-> TxMeasure blk
-> LedgerTables blk ValuesMK
-> TickedLedgerState blk ValuesMK
-> InternalState blk
-> (Either
      (ApplyTxErr blk) (Validated (GenTx blk), LedgerTables blk DiffMK),
    DiffTimeMeasure -> InternalState blk)
forall blk.
(LedgerSupportsMempool blk, HasTxId (GenTx blk)) =>
LedgerConfig blk
-> WhetherToIntervene
-> GenTx blk
-> TxMeasure blk
-> LedgerTables blk ValuesMK
-> TickedLedgerState blk ValuesMK
-> InternalState blk
-> (Either
      (ApplyTxErr blk) (Validated (GenTx blk), LedgerTables blk DiffMK),
    DiffTimeMeasure -> InternalState blk)
validateNewTransaction LedgerCfg LedgerState blk
cfg WhetherToIntervene
wti GenTx blk
tx (TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
forall blk.
TxMeasurePhase1 blk -> TxMeasurePhase2 blk -> TxMeasure blk
TxMeasure TxMeasurePhase1 blk
p1TxMeasure TxMeasurePhase2 blk
txsz) LedgerTables blk ValuesMK
values TickedLedgerState blk ValuesMK
st InternalState blk
is of
                (Left ApplyTxErr blk
err, DiffTimeMeasure -> InternalState blk
_) ->
                  (DiffTimeMeasure -> TransactionProcessed blk) -> TriedToAddTx blk
forall blk.
(DiffTimeMeasure -> TransactionProcessed blk) -> TriedToAddTx blk
Processed ((DiffTimeMeasure -> TransactionProcessed blk) -> TriedToAddTx blk)
-> (DiffTimeMeasure -> TransactionProcessed blk)
-> TriedToAddTx blk
forall a b. (a -> b) -> a -> b
$ \DiffTimeMeasure
_dur ->
                    Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
forall blk.
Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
TransactionProcessingResult
                      Maybe (InternalState blk)
forall a. Maybe a
Nothing
                      (GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
forall blk. GenTx blk -> ApplyTxErr blk -> MempoolAddTxResult blk
MempoolTxRejected GenTx blk
tx ApplyTxErr blk
err)
                      ( GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
forall blk.
GenTx blk
-> ApplyTxErr blk
-> MempoolRejectionDetails
-> MempoolSize
-> TraceEventMempool blk
TraceMempoolRejectedTx
                          GenTx blk
tx
                          ApplyTxErr blk
err
                          MempoolRejectionDetails
MempoolRejectedByLedger
                          (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is)
                      )
                (Right (Validated (GenTx blk)
vtx, LedgerTables blk DiffMK
df), DiffTimeMeasure -> InternalState blk
is') ->
                  (DiffTimeMeasure -> TransactionProcessed blk) -> TriedToAddTx blk
forall blk.
(DiffTimeMeasure -> TransactionProcessed blk) -> TriedToAddTx blk
Processed ((DiffTimeMeasure -> TransactionProcessed blk) -> TriedToAddTx blk)
-> (DiffTimeMeasure -> TransactionProcessed blk)
-> TriedToAddTx blk
forall a b. (a -> b) -> a -> b
$ \DiffTimeMeasure
dur ->
                    Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
forall blk.
Maybe (InternalState blk)
-> MempoolAddTxResult blk
-> TraceEventMempool blk
-> TransactionProcessed blk
TransactionProcessingResult
                      (InternalState blk -> Maybe (InternalState blk)
forall a. a -> Maybe a
Just (DiffTimeMeasure -> InternalState blk
is' DiffTimeMeasure
dur))
                      (Validated (GenTx blk)
-> LedgerTables blk DiffMK -> MempoolAddTxResult blk
forall blk.
Validated (GenTx blk)
-> LedgerTables blk DiffMK -> MempoolAddTxResult blk
MempoolTxAdded Validated (GenTx blk)
vtx LedgerTables blk DiffMK
df)
                      ( Validated (GenTx blk)
-> MempoolSize -> MempoolSize -> TraceEventMempool blk
forall blk.
Validated (GenTx blk)
-> MempoolSize -> MempoolSize -> TraceEventMempool blk
TraceMempoolAddedTx
                          Validated (GenTx blk)
vtx
                          (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize InternalState blk
is)
                          (InternalState blk -> MempoolSize
forall blk. TxLimits blk => InternalState blk -> MempoolSize
isMempoolSize (DiffTimeMeasure -> InternalState blk
is' DiffTimeMeasure
dur))
                      )
 where
  currentSize :: TxMeasureWithDiffTime blk
currentSize = TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> TxMeasureWithDiffTime blk
forall sz tx. Measure sz => TxSeq sz tx -> sz
TxSeq.toSize (InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is)

{-------------------------------------------------------------------------------
  Remove transactions
-------------------------------------------------------------------------------}

-- | See 'Ouroboros.Consensus.Mempool.API.removeTxsEvenIfValid'.
implRemoveTxsEvenIfValid ::
  ( IOLike m
  , LedgerSupportsMempool blk
  , HasTxId (GenTx blk)
  ) =>
  MempoolEnv m blk ->
  NE.NonEmpty (GenTxId blk) ->
  m ()
implRemoveTxsEvenIfValid :: forall (m :: * -> *) blk.
(IOLike m, LedgerSupportsMempool blk, HasTxId (GenTx blk)) =>
MempoolEnv m blk -> NonEmpty (GenTxId blk) -> m ()
implRemoveTxsEvenIfValid MempoolEnv m blk
mpEnv NonEmpty (GenTxId blk)
toRemove =
  StrictTMVar m (InternalState blk)
-> (InternalState blk -> m ((), InternalState blk)) -> m ()
forall (m :: * -> *) a c.
IOLike m =>
StrictTMVar m a -> (a -> m (c, a)) -> m c
withTMVar StrictTMVar m (InternalState blk)
istate ((InternalState blk -> m ((), InternalState blk)) -> m ())
-> (InternalState blk -> m ((), InternalState blk)) -> m ()
forall a b. (a -> b) -> a -> b
$
    \InternalState blk
is -> do
      let toKeep :: [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
toKeep =
            (TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
 -> Bool)
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall a. (a -> Bool) -> [a] -> [a]
filter
              ( (GenTxId blk -> Set (GenTxId blk) -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`notElem` [GenTxId blk] -> Set (GenTxId blk)
forall a. Ord a => [a] -> Set a
Set.fromList (NonEmpty (GenTxId blk) -> [GenTxId blk]
forall a. NonEmpty a -> [a]
NE.toList NonEmpty (GenTxId blk)
toRemove))
                  (GenTxId blk -> Bool)
-> (TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
    -> GenTxId blk)
-> TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. GenTx blk -> GenTxId blk
forall tx. HasTxId tx => tx -> TxId tx
txId
                  (GenTx blk -> GenTxId blk)
-> (TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
    -> GenTx blk)
-> TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> GenTxId blk
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Validated (GenTx blk) -> GenTx blk
forall blk.
LedgerSupportsMempool blk =>
Validated (GenTx blk) -> GenTx blk
txForgetValidated
                  (Validated (GenTx blk) -> GenTx blk)
-> (TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
    -> Validated (GenTx blk))
-> TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> GenTx blk
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ValidatedTxWithDiffs blk -> Validated (GenTx blk)
forall blk. ValidatedTxWithDiffs blk -> Validated (GenTx blk)
validatedTx
                  (ValidatedTxWithDiffs blk -> Validated (GenTx blk))
-> (TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
    -> ValidatedTxWithDiffs blk)
-> TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> Validated (GenTx blk)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> ValidatedTxWithDiffs blk
forall sz tx. TxTicket sz tx -> tx
txTicketTx
              )
              (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall sz tx. TxSeq sz tx -> [TxTicket sz tx]
TxSeq.toList (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
 -> [TxTicket
       (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)])
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall a b. (a -> b) -> a -> b
$ InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is)
          -- A tx actually leaves iff we kept fewer than we had; bump the removal
          -- generation then, so an in-flight sync notices its snapshot is stale
          -- (see 'isRemovalCounter'). Carried on the state itself, under the lock.
          newGen :: Word64
newGen
            | [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
-> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
toKeep Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk) -> Int
forall a. TxSeq (TxMeasureWithDiffTime blk) a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length (InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is) = InternalState blk -> Word64
forall blk. InternalState blk -> Word64
isRemovalCounter InternalState blk
is Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ Word64
1
            | Bool
otherwise = InternalState blk -> Word64
forall blk. InternalState blk -> Word64
isRemovalCounter InternalState blk
is
      frkr <- StrictMVar m (ReadOnlyForker m LedgerState blk)
-> m (ReadOnlyForker m LedgerState blk)
forall (m :: * -> *) a. MonadMVar m => StrictMVar m a -> m a
readMVar StrictMVar m (ReadOnlyForker m LedgerState blk)
forker
      RevalidateTxsResult is' removed <-
        revalidateTxsFor
          frkr
          capacityOverride
          cfg
          (isSlotNo is)
          (isLedgerState is `withLedgerTables` emptyLedgerTables)
          (isLastTicketNo is)
          newGen
          toKeep
      traceWith trcr $
        TraceMempoolManuallyRemovedTxs toRemove (map getInvalidated removed) (isMempoolSize is')
      pure ((), is')
 where
  MempoolEnv
    { mpEnvStateVar :: forall (m :: * -> *) blk.
MempoolEnv m blk -> StrictTMVar m (InternalState blk)
mpEnvStateVar = StrictTMVar m (InternalState blk)
istate
    , mpEnvForker :: forall (m :: * -> *) blk.
MempoolEnv m blk -> StrictMVar m (ReadOnlyForker m LedgerState blk)
mpEnvForker = StrictMVar m (ReadOnlyForker m LedgerState blk)
forker
    , mpEnvTracer :: forall (m :: * -> *) blk.
MempoolEnv m blk -> Tracer m (TraceEventMempool blk)
mpEnvTracer = Tracer m (TraceEventMempool blk)
trcr
    , mpEnvLedgerCfg :: forall (m :: * -> *) blk. MempoolEnv m blk -> LedgerConfig blk
mpEnvLedgerCfg = LedgerConfig blk
cfg
    , mpEnvCapacityOverride :: forall (m :: * -> *) blk.
MempoolEnv m blk -> MempoolCapacityBytesOverride
mpEnvCapacityOverride = MempoolCapacityBytesOverride
capacityOverride
    } = MempoolEnv m blk
mpEnv

{-------------------------------------------------------------------------------
  Sync with ledger
-------------------------------------------------------------------------------}

-- | Maximum number of delta transactions 'implSyncWithLedger' will reapply
-- while holding the state lock. It shrinks its outstanding delta off the lock
-- until at most this many txs remain, so the final under-lock reapply — and
-- hence the time snapshot readers can be blocked — is bounded independently of
-- mempool occupancy.
syncDeltaCap :: Int
syncDeltaCap :: Int
syncDeltaCap = Int
256

-- | Safety valve on the off-lock shrink loop: if the delta never falls below
-- 'syncDeltaCap' (e.g. reapplication is not actually cheaper than ingestion),
-- stop after this many rounds and finish anyway, degrading to a larger
-- under-lock reapply rather than looping unboundedly.
syncMaxIters :: Int
syncMaxIters :: Int
syncMaxIters = Int
8

-- | See 'Ouroboros.Consensus.Mempool.API.syncWithLedger'
implSyncWithLedger ::
  forall m blk r.
  ( IOLike m
  , LedgerSupportsMempool blk
  , ValidateEnvelope blk
  , HasTxId (GenTx blk)
  ) =>
  -- | This argument is only to be able to acquire a snapshot in the same
  -- atomically block as the re-sync when testing the mempool in the QSM
  -- parallel tests. We could instead always compute a snapshot and ignore it in
  -- the common case, but it seems acceptable to not even create the thunk for
  -- it. This will be set to @const ()@ on the code that is run by the node.
  (InternalState blk -> r) ->
  MempoolEnv m blk ->
  m r
implSyncWithLedger :: forall (m :: * -> *) blk r.
(IOLike m, LedgerSupportsMempool blk, ValidateEnvelope blk,
 HasTxId (GenTx blk)) =>
(InternalState blk -> r) -> MempoolEnv m blk -> m r
implSyncWithLedger InternalState blk -> r
projectResult MempoolEnv m blk
mpEnv =
  Tracer m EnclosingTimed -> m r -> m r
forall (m :: * -> *) a.
MonadMonotonicTime m =>
Tracer m EnclosingTimed -> m a -> m a
encloseTimedWith (EnclosingTimed -> TraceEventMempool blk
forall blk. EnclosingTimed -> TraceEventMempool blk
TraceMempoolSynced (EnclosingTimed -> TraceEventMempool blk)
-> Tracer m (TraceEventMempool blk) -> Tracer m EnclosingTimed
forall (f :: * -> *) a b. Contravariant f => (a -> b) -> f b -> f a
>$< MempoolEnv m blk -> Tracer m (TraceEventMempool blk)
forall (m :: * -> *) blk.
MempoolEnv m blk -> Tracer m (TraceEventMempool blk)
mpEnvTracer MempoolEnv m blk
mpEnv) m r
goSync
 where
  -- Sync with a bounded lock hold. Everything expensive is done /off the lock/,
  -- against a snapshot @is0@ taken with a non-emptying 'readTMVar' while adds
  -- keep appending: the big LedgerDB read of @is0@'s inputs, the revalidation of
  -- @is0@'s txs at the new tip (@cand0@), and then a converging loop that reads
  -- the txs added in the meantime (the "delta", by 'TicketNo') and reapplies
  -- them /on top/ via 'revalidateTxsFor'' — never reprocessing what is already done.
  --
  -- The delta shrinks each iteration: adds are serialised (the fifo 'MVar's) and
  -- pay full validation, whereas the sync only reapplies, which is cheaper per
  -- tx, so fewer txs arrive during a round than it processes. Once the delta is
  -- at most 'syncDeltaCap' (or we hit 'syncMaxIters'), we 'takeTMVar' and
  -- reapply just that bounded residual before swapping. So the lock is held for
  -- a near-constant time — O('syncDeltaCap') — independent of mempool occupancy,
  -- rather than for a full O(n) revalidation.
  goSync :: m r
goSync =
    m (Either
     (InternalState blk)
     (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
      InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
      ChainHash (LedgerState blk)))
checkTodo m (Either
     (InternalState blk)
     (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
      InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
      ChainHash (LedgerState blk)))
-> (Either
      (InternalState blk)
      (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
       InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
       ChainHash (LedgerState blk))
    -> m r)
-> m r
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Left InternalState blk
is0 -> do
        -- The tip didn't change, put the same state.
        Tracer m (TraceEventMempool blk) -> TraceEventMempool blk -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceEventMempool blk)
trcr (TraceEventMempool blk -> m ()) -> TraceEventMempool blk -> m ()
forall a b. (a -> b) -> a -> b
$ Point blk -> TraceEventMempool blk
forall blk. Point blk -> TraceEventMempool blk
TraceMempoolSyncNotNeeded (InternalState blk -> Point blk
forall blk. InternalState blk -> Point blk
isTip InternalState blk
is0)
        r -> m r
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (InternalState blk -> r
projectResult InternalState blk
is0)
      Right (m (Either GetForkerError (ReadOnlyForker m LedgerState blk))
getForker, InternalState blk
is0, SlotNo
slot, Ticked LedgerState blk DiffMK
ls, ChainHash (LedgerState blk)
tipHash0) ->
        -- The tip changed, we have to revalidate.
        -- NOTE: The forker is closed in a bracket here because exceptions here
        -- are fatal anyways. See also 'mldViewGetForker' (the function we call
        -- here).
        m (Either GetForkerError (ReadOnlyForker m LedgerState blk))
getForker m (Either GetForkerError (ReadOnlyForker m LedgerState blk))
-> (Either GetForkerError (ReadOnlyForker m LedgerState blk)
    -> m r)
-> m r
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          -- This case should happen only if the tip has moved again, this time
          -- to a separate fork, since the background thread saw a change in the
          -- tip, which should happen very rarely
          Left{} -> do
            Tracer m (TraceEventMempool blk) -> TraceEventMempool blk -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m (TraceEventMempool blk)
trcr TraceEventMempool blk
forall blk. TraceEventMempool blk
TraceMempoolTipMovedBetweenSTMBlocks
            m r
goSync
          Right ReadOnlyForker m LedgerState blk
frk -> do
            -- OFF-LOCK: read the snapshot's inputs and revalidate the entire
            -- mempool against the new tip.
            seed <-
              ReadOnlyForker m LedgerState blk
-> MempoolCapacityBytesOverride
-> LedgerConfig blk
-> SlotNo
-> Ticked LedgerState blk DiffMK
-> TicketNo
-> Word64
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
-> m (RevalidateTxsResult blk)
forall (m :: * -> *) blk.
(Monad m, LedgerSupportsMempool blk, HasTxId (GenTx blk)) =>
ReadOnlyForker m LedgerState blk
-> MempoolCapacityBytesOverride
-> LedgerConfig blk
-> SlotNo
-> TickedLedgerState blk DiffMK
-> TicketNo
-> Word64
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
-> m (RevalidateTxsResult blk)
revalidateTxsFor
                ReadOnlyForker m LedgerState blk
frk
                MempoolCapacityBytesOverride
capacityOverride
                LedgerConfig blk
cfg
                SlotNo
slot
                Ticked LedgerState blk DiffMK
ls
                (InternalState blk -> TicketNo
forall blk. InternalState blk -> TicketNo
isLastTicketNo InternalState blk
is0)
                (InternalState blk -> Word64
forall blk. InternalState blk -> Word64
isRemovalCounter InternalState blk
is0)
                (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall sz tx. TxSeq sz tx -> [TxTicket sz tx]
TxSeq.toList (InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is0))
            -- A bounded catch-up loop reapplying whatever was added or
            -- removed while this ran, so the state lock is held only briefly at
            -- the very end.
            revalidateDeltas frk tipHash0 slot seed (0 :: Int) >>= \case
              Maybe r
Nothing -> m r
goSync -- Retry if the commit found the state stale.
              Just r
r -> r -> m r
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure r
r

  checkTodo :: m (Either
     (InternalState blk)
     (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
      InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
      ChainHash (LedgerState blk)))
checkTodo = STM
  m
  (Either
     (InternalState blk)
     (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
      InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
      ChainHash (LedgerState blk)))
-> m (Either
        (InternalState blk)
        (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
         InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
         ChainHash (LedgerState blk)))
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM
   m
   (Either
      (InternalState blk)
      (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
       InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
       ChainHash (LedgerState blk)))
 -> m (Either
         (InternalState blk)
         (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
          InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
          ChainHash (LedgerState blk))))
-> STM
     m
     (Either
        (InternalState blk)
        (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
         InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
         ChainHash (LedgerState blk)))
-> m (Either
        (InternalState blk)
        (m (Either GetForkerError (ReadOnlyForker m LedgerState blk)),
         InternalState blk, SlotNo, Ticked LedgerState blk DiffMK,
         ChainHash (LedgerState blk)))
forall a b. (a -> b) -> a -> b
$ do
    view <- LedgerInterface m blk -> STM m (MempoolLedgerDBView m blk)
forall (m :: * -> *) blk.
LedgerInterface m blk -> STM m (MempoolLedgerDBView m blk)
getCurrentLedgerState LedgerInterface m blk
ldgrInterface
    is0 <- readTMVar istate
    let ls0 = MempoolLedgerDBView m blk -> LedgerState blk EmptyMK
forall (m :: * -> *) blk.
MempoolLedgerDBView m blk -> LedgerState blk EmptyMK
mldViewState MempoolLedgerDBView m blk
view
        tipHash0 = LedgerState blk EmptyMK -> ChainHash (LedgerState blk)
forall (l :: MapKind -> *) (mk :: MapKind).
GetTip l =>
l mk -> ChainHash l
getTipHash LedgerState blk EmptyMK
ls0
        (slot, ls') = tickLedgerState cfg $ ForgeInUnknownSlot ls0
    pure $
      if pointHash (isTip is0) == castHash tipHash0 && isSlotNo is0 == slot
        then Left is0
        else Right (mldViewGetForker view, is0, slot, ls', tipHash0)

  deltaAfter :: InternalState blk
-> InternalState blk
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
deltaAfter InternalState blk
cand InternalState blk
is =
    TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall sz tx. TxSeq sz tx -> [TxTicket sz tx]
TxSeq.toList (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
 -> [TxTicket
       (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)])
-> ((TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk),
     TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
    -> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
-> (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk),
    TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk),
 TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall a b. (a, b) -> b
snd ((TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk),
  TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
 -> [TxTicket
       (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)])
-> (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk),
    TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall a b. (a -> b) -> a -> b
$ TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
-> TicketNo
-> (TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk),
    TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk))
forall sz tx.
Measure sz =>
TxSeq sz tx -> TicketNo -> (TxSeq sz tx, TxSeq sz tx)
TxSeq.splitAfterTicketNo (InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
forall blk.
InternalState blk
-> TxSeq (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)
isTxs InternalState blk
is) (InternalState blk -> TicketNo
forall blk. InternalState blk -> TicketNo
isLastTicketNo InternalState blk
cand)

  -- Revalidate the outstanding deltas off the lock until small enough, then
  -- finish under it with a bounded residual reapply. Returns Nothing (retry the
  -- whole sync) if the tip moved or a tx was removed while we worked.
  -- NOTE: Either closes the forker or updates it in the InternalState
  revalidateDeltas ::
    ReadOnlyForker m LedgerState blk ->
    ChainHash (LedgerState blk) ->
    SlotNo ->
    RevalidateTxsResult blk ->
    Int ->
    m (Maybe r)
  revalidateDeltas :: ReadOnlyForker m LedgerState blk
-> ChainHash (LedgerState blk)
-> SlotNo
-> RevalidateTxsResult blk
-> Int
-> m (Maybe r)
revalidateDeltas ReadOnlyForker m LedgerState blk
frk ChainHash (LedgerState blk)
tipHash0 SlotNo
slot = RevalidateTxsResult blk -> Int -> m (Maybe r)
go
   where
    -- The candidate is doomed the moment the tip moves or a tx is force-removed
    -- (the commit below would reject it, retrying the whole sync). This helper
    -- checks that condition, run on each iteration.
    stale :: InternalState blk
-> InternalState blk -> ChainHash (LedgerState blk) -> Bool
stale InternalState blk
cand InternalState blk
isNow ChainHash (LedgerState blk)
curTipHash =
      ChainHash (LedgerState blk)
curTipHash ChainHash (LedgerState blk) -> ChainHash (LedgerState blk) -> Bool
forall a. Eq a => a -> a -> Bool
/= ChainHash (LedgerState blk)
tipHash0 Bool -> Bool -> Bool
|| InternalState blk -> Word64
forall blk. InternalState blk -> Word64
isRemovalCounter InternalState blk
isNow Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
/= InternalState blk -> Word64
forall blk. InternalState blk -> Word64
isRemovalCounter InternalState blk
cand

    go :: RevalidateTxsResult blk -> Int -> m (Maybe r)
go acc :: RevalidateTxsResult blk
acc@(RevalidateTxsResult InternalState blk
cand [Invalidated blk]
_) Int
iterN = do
      (isNow, curTipHash) <- STM m (InternalState blk, ChainHash (LedgerState blk))
-> m (InternalState blk, ChainHash (LedgerState blk))
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (InternalState blk, ChainHash (LedgerState blk))
 -> m (InternalState blk, ChainHash (LedgerState blk)))
-> STM m (InternalState blk, ChainHash (LedgerState blk))
-> m (InternalState blk, ChainHash (LedgerState blk))
forall a b. (a -> b) -> a -> b
$ do
        isNow <- StrictTMVar m (InternalState blk) -> STM m (InternalState blk)
forall (m :: * -> *) a. MonadSTM m => StrictTMVar m a -> STM m a
readTMVar StrictTMVar m (InternalState blk)
istate
        view <- getCurrentLedgerState ldgrInterface
        pure (isNow, getTipHash (mldViewState view))
      if stale cand isNow curTipHash
        then do
          -- Off-lock early abort: 'goSync' will retry from a fresh snapshot.
          roforkerClose frk
          pure Nothing
        else do
          let deltaTickets = InternalState blk
-> InternalState blk
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall {blk} {blk}.
(Measure (TxMeasurePhase1 blk), Measure (TxMeasurePhase2 blk)) =>
InternalState blk
-> InternalState blk
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
deltaAfter InternalState blk
cand InternalState blk
isNow
          if length deltaTickets > syncDeltaCap && iterN < syncMaxIters
            then do
              next <- revalidateTxsFor' frk capacityOverride cfg slot acc (isLastTicketNo isNow) deltaTickets
              go next (iterN + 1)
            else
              -- Delta is small enough (or out of iterations); commit under the lock.
              withTMVarAnd istate (const $ getCurrentLedgerState ldgrInterface) $
                \InternalState blk
isLocked (MempoolLedgerDBView LedgerState blk EmptyMK
ls m (Either GetForkerError (ReadOnlyForker m LedgerState blk))
_getForker) -> do
                  -- ON-LOCK: the same staleness check, now atomic with the commit.
                  -- The off-lock checks above cannot be final: the tip can still
                  -- move between the last one and acquiring the lock. Re-reading
                  -- the ledger state here (in the STM transaction that takes the
                  -- lock) is what makes the tip-moved check atomic with the swap;
                  -- committing a stale candidate would resurrect a dropped tx or
                  -- publish a snapshot for a tip that is no longer current, i.e. be
                  -- non-linearizable.
                  if InternalState blk
-> InternalState blk -> ChainHash (LedgerState blk) -> Bool
stale InternalState blk
cand InternalState blk
isLocked (LedgerState blk EmptyMK -> ChainHash (LedgerState blk)
forall (l :: MapKind -> *) (mk :: MapKind).
GetTip l =>
l mk -> ChainHash l
getTipHash LedgerState blk EmptyMK
ls)
                    then do
                      -- As we won't be keeping the forker in the internal state, we
                      -- must close it.
                      ReadOnlyForker m LedgerState blk -> m ()
forall (m :: * -> *) (l :: * -> MapKind -> *) blk.
ReadOnlyForker m l blk -> m ()
roforkerClose ReadOnlyForker m LedgerState blk
frk
                      (Maybe r, InternalState blk) -> m (Maybe r, InternalState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe r
forall a. Maybe a
Nothing, InternalState blk
isLocked)
                    else do
                      -- Lock held, so no add can intervene: reapply just the residual
                      -- delta (the cap plus stragglers that landed while acquiring it).
                      let resTickets :: [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
resTickets = InternalState blk
-> InternalState blk
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
forall {blk} {blk}.
(Measure (TxMeasurePhase1 blk), Measure (TxMeasurePhase2 blk)) =>
InternalState blk
-> InternalState blk
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
deltaAfter InternalState blk
cand InternalState blk
isLocked
                      RevalidateTxsResult isFinal removed <-
                        ReadOnlyForker m LedgerState blk
-> MempoolCapacityBytesOverride
-> LedgerConfig blk
-> SlotNo
-> RevalidateTxsResult blk
-> TicketNo
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
-> m (RevalidateTxsResult blk)
forall (m :: * -> *) blk.
(Monad m, LedgerSupportsMempool blk, HasTxId (GenTx blk)) =>
ReadOnlyForker m LedgerState blk
-> MempoolCapacityBytesOverride
-> LedgerConfig blk
-> SlotNo
-> RevalidateTxsResult blk
-> TicketNo
-> [TxTicket
      (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
-> m (RevalidateTxsResult blk)
revalidateTxsFor' ReadOnlyForker m LedgerState blk
frk MempoolCapacityBytesOverride
capacityOverride LedgerConfig blk
cfg SlotNo
slot RevalidateTxsResult blk
acc (InternalState blk -> TicketNo
forall blk. InternalState blk -> TicketNo
isLastTicketNo InternalState blk
isLocked) [TxTicket (TxMeasureWithDiffTime blk) (ValidatedTxWithDiffs blk)]
resTickets
                      unless (null removed) $
                        traceWith trcr $
                          TraceMempoolRemoveTxs
                            (map (\Invalidated blk
x -> (Invalidated blk -> GenTx blk
forall blk. Invalidated blk -> GenTx blk
getInvalidated Invalidated blk
x, Invalidated blk -> ApplyTxErr blk
forall blk. Invalidated blk -> ApplyTxErr blk
getReason Invalidated blk
x)) removed)
                            (isMempoolSize isFinal)
                      -- Store the forker to be used with the new state
                      modifyMVar_ forkerMVar (\ReadOnlyForker m LedgerState blk
frkOld -> ReadOnlyForker m LedgerState blk -> m ()
forall (m :: * -> *) (l :: * -> MapKind -> *) blk.
ReadOnlyForker m l blk -> m ()
roforkerClose ReadOnlyForker m LedgerState blk
frkOld m ()
-> m (ReadOnlyForker m LedgerState blk)
-> m (ReadOnlyForker m LedgerState blk)
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> ReadOnlyForker m LedgerState blk
-> m (ReadOnlyForker m LedgerState blk)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ReadOnlyForker m LedgerState blk
frk)
                      pure (Just (projectResult isFinal), isFinal)

  MempoolEnv
    { mpEnvStateVar :: forall (m :: * -> *) blk.
MempoolEnv m blk -> StrictTMVar m (InternalState blk)
mpEnvStateVar = StrictTMVar m (InternalState blk)
istate
    , mpEnvForker :: forall (m :: * -> *) blk.
MempoolEnv m blk -> StrictMVar m (ReadOnlyForker m LedgerState blk)
mpEnvForker = StrictMVar m (ReadOnlyForker m LedgerState blk)
forkerMVar
    , mpEnvLedger :: forall (m :: * -> *) blk. MempoolEnv m blk -> LedgerInterface m blk
mpEnvLedger = LedgerInterface m blk
ldgrInterface
    , mpEnvTracer :: forall (m :: * -> *) blk.
MempoolEnv m blk -> Tracer m (TraceEventMempool blk)
mpEnvTracer = Tracer m (TraceEventMempool blk)
trcr
    , mpEnvLedgerCfg :: forall (m :: * -> *) blk. MempoolEnv m blk -> LedgerConfig blk
mpEnvLedgerCfg = LedgerConfig blk
cfg
    , mpEnvCapacityOverride :: forall (m :: * -> *) blk.
MempoolEnv m blk -> MempoolCapacityBytesOverride
mpEnvCapacityOverride = MempoolCapacityBytesOverride
capacityOverride
    } = MempoolEnv m blk
mpEnv