{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE ScopedTypeVariables #-}

module Cardano.Tools.DBSynthesizer.Forging
  ( GenTxs
  , runForge
  ) where

import Cardano.Tools.DBSynthesizer.Types
  ( ForgeLimit (..)
  , ForgeResult (..)
  )
import Control.Monad (unless, void)
import Control.Tracer (nullTracer)
import Data.Maybe (isJust)
import Data.Word (Word64)
import Ouroboros.Consensus.Block
  ( BlockForging
  , EpochSize (..)
  , SlotNo (..)
  )
import Ouroboros.Consensus.Config
  ( TopLevelConfig
  , configLedger
  )
import Ouroboros.Consensus.Ledger.Abstract
  ( ComputeLedgerEvents (..)
  , DiffMK
  , TickedLedgerState
  , applyChainTick
  )
import Ouroboros.Consensus.Ledger.Extended (ledgerState)
import Ouroboros.Consensus.Ledger.SupportsMempool
  ( GenTx
  )
import Ouroboros.Consensus.Mempool
  ( Mempool (..)
  , MempoolAddTxResult (MempoolTxRejected)
  , MempoolCapacityBytesOverride (..)
  , addLocalTxs
  , chainDBLedgerInterface
  , openMempoolWithoutSyncThread
  )
import Ouroboros.Consensus.Node.Run (RunNode)
import Ouroboros.Consensus.NodeKernel.Forge (forge)
import Ouroboros.Consensus.Storage.ChainDB.API as ChainDB
  ( ChainDB
  , getTipPoint
  , withReadOnlyForkerAtPoint
  )
import Ouroboros.Consensus.Storage.LedgerDB
  ( ReadOnlyForker'
  , roforkerGetLedgerState
  )
import Ouroboros.Consensus.Util.EarlyExit
  ( exitEarly
  , lift
  , withEarlyExit
  )
import Ouroboros.Consensus.Util.IOLike (atomically)
import Ouroboros.Network.Protocol.LocalStateQuery.Type (Target (..))
import System.Timeout (timeout)

data ForgeState
  = ForgeState
  { ForgeState -> SlotNo
currentSlot :: !SlotNo
  , ForgeState -> Word64
forged :: !Word64
  , ForgeState -> Word64
currentEpoch :: !Word64
  , ForgeState -> SlotNo
processed :: !SlotNo
  }

initialForgeState :: ForgeState
initialForgeState :: ForgeState
initialForgeState = SlotNo -> Word64 -> Word64 -> SlotNo -> ForgeState
ForgeState SlotNo
0 Word64
0 Word64
0 SlotNo
0

-- | An action to generate transactions for a given block
-- If Mempool capacity (2 blocks) is breached with generated transactions,
-- 'Mempool.addLocalTxs' blocks until a specified timeout.
type GenTxs blk =
  SlotNo ->
  ReadOnlyForker' IO blk ->
  TickedLedgerState blk DiffMK ->
  IO [GenTx blk]

-- In microseconds
mempoolAddTxsTimeout :: Int
mempoolAddTxsTimeout :: Int
mempoolAddTxsTimeout = Int
100_000 -- 0.1s

runForge ::
  forall blk.
  RunNode blk =>
  EpochSize ->
  SlotNo ->
  ForgeLimit ->
  ChainDB IO blk ->
  [BlockForging IO blk] ->
  TopLevelConfig blk ->
  GenTxs blk ->
  IO ForgeResult
runForge :: forall blk.
RunNode blk =>
EpochSize
-> SlotNo
-> ForgeLimit
-> ChainDB IO blk
-> [BlockForging IO blk]
-> TopLevelConfig blk
-> GenTxs blk
-> IO ForgeResult
runForge EpochSize
epochSize_ SlotNo
nextSlot ForgeLimit
opts ChainDB IO blk
chainDB [BlockForging IO blk]
blockForging TopLevelConfig blk
cfg GenTxs blk
genTxs = do
  String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"--> epoch size: " String -> String -> String
forall a. [a] -> [a] -> [a]
++ EpochSize -> String
forall a. Show a => a -> String
show EpochSize
epochSize_
  String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"--> will process until: " String -> String -> String
forall a. [a] -> [a] -> [a]
++ ForgeLimit -> String
forall a. Show a => a -> String
show ForgeLimit
opts
  -- TODO(bladyjoker): I think it would be simpler to just have the sync thread.
  mempool <-
    LedgerInterface IO blk
-> LedgerCfg LedgerState blk
-> MempoolCapacityBytesOverride
-> Maybe MempoolTimeoutConfig
-> Tracer IO (TraceEventMempool blk)
-> IO (Mempool IO blk)
forall (m :: * -> *) blk.
(IOLike m, MonadTimer m, LedgerSupportsMempool blk,
 HasTxId (GenTx blk), ValidateEnvelope blk) =>
LedgerInterface m blk
-> LedgerConfig blk
-> MempoolCapacityBytesOverride
-> Maybe MempoolTimeoutConfig
-> Tracer m (TraceEventMempool blk)
-> m (Mempool m blk)
openMempoolWithoutSyncThread
      (ChainDB IO blk -> LedgerInterface IO blk
forall (m :: * -> *) blk.
(IOLike m, IsLedger LedgerState blk) =>
ChainDB m blk -> LedgerInterface m blk
chainDBLedgerInterface ChainDB IO blk
chainDB)
      (TopLevelConfig blk -> LedgerCfg LedgerState blk
forall blk. TopLevelConfig blk -> LedgerConfig blk
configLedger TopLevelConfig blk
cfg)
      MempoolCapacityBytesOverride
NoMempoolCapacityBytesOverride
      Maybe MempoolTimeoutConfig
forall a. Maybe a
Nothing
      Tracer IO (TraceEventMempool blk)
forall (m :: * -> *) a. Monad m => Tracer m a
nullTracer
  endState <- go mempool initialForgeState{currentSlot = nextSlot}
  putStrLn $
    "--> forged and adopted "
      ++ show (forged endState)
      ++ " blocks; reached "
      ++ show (currentSlot endState)
  pure $ ForgeResult $ fromIntegral $ forged endState
 where
  epochSize :: Word64
epochSize = EpochSize -> Word64
unEpochSize EpochSize
epochSize_

  forgingDone :: ForgeState -> Bool
  forgingDone :: ForgeState -> Bool
forgingDone = case ForgeLimit
opts of
    ForgeLimitSlot SlotNo
s -> (SlotNo
s SlotNo -> SlotNo -> Bool
forall a. Eq a => a -> a -> Bool
==) (SlotNo -> Bool) -> (ForgeState -> SlotNo) -> ForgeState -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ForgeState -> SlotNo
processed
    ForgeLimitBlock Word64
b -> (Word64
b Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
==) (Word64 -> Bool) -> (ForgeState -> Word64) -> ForgeState -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ForgeState -> Word64
forged
    ForgeLimitEpoch Word64
e -> (Word64
e Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
==) (Word64 -> Bool) -> (ForgeState -> Word64) -> ForgeState -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ForgeState -> Word64
currentEpoch

  go :: Mempool IO blk -> ForgeState -> IO ForgeState
  go :: Mempool IO blk -> ForgeState -> IO ForgeState
go Mempool IO blk
mempool ForgeState
forgeState
    | ForgeState -> Bool
forgingDone ForgeState
forgeState = ForgeState -> IO ForgeState
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ForgeState
forgeState
    | Bool
otherwise = do
        didForge <- Mempool IO blk -> SlotNo -> IO Bool
goSlot Mempool IO blk
mempool (ForgeState -> SlotNo
currentSlot ForgeState
forgeState)
        go mempool (nextForgeState forgeState didForge)

  nextForgeState :: ForgeState -> Bool -> ForgeState
  nextForgeState :: ForgeState -> Bool -> ForgeState
nextForgeState ForgeState{SlotNo
currentSlot :: ForgeState -> SlotNo
currentSlot :: SlotNo
currentSlot, Word64
forged :: ForgeState -> Word64
forged :: Word64
forged, Word64
currentEpoch :: ForgeState -> Word64
currentEpoch :: Word64
currentEpoch, SlotNo
processed :: ForgeState -> SlotNo
processed :: SlotNo
processed} Bool
didForge =
    ForgeState
      { currentSlot :: SlotNo
currentSlot = SlotNo
currentSlot SlotNo -> SlotNo -> SlotNo
forall a. Num a => a -> a -> a
+ SlotNo
1
      , forged :: Word64
forged = Word64
forged Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ if Bool
didForge then Word64
1 else Word64
0
      , currentEpoch :: Word64
currentEpoch = Word64
epoch'
      , processed :: SlotNo
processed = SlotNo
processed'
      }
   where
    processed' :: SlotNo
processed' = SlotNo
processed SlotNo -> SlotNo -> SlotNo
forall a. Num a => a -> a -> a
+ SlotNo
1
    epoch' :: Word64
epoch' = Word64
currentEpoch Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ if SlotNo -> Word64
unSlotNo SlotNo
processed' Word64 -> Word64 -> Word64
forall a. Integral a => a -> a -> a
`rem` Word64
epochSize Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
== Word64
0 then Word64
1 else Word64
0

  goSlot :: Mempool IO blk -> SlotNo -> IO Bool
  goSlot :: Mempool IO blk -> SlotNo -> IO Bool
goSlot Mempool IO blk
mempool SlotNo
currentSlot = do
    -- Mempool and ChainDB must see the same 'tip'
    -- Sync mempool with the latest block adoption at 'tip'
    _ <- Mempool IO blk -> IO (MempoolSnapshot blk)
forall (m :: * -> *) blk. Mempool m blk -> m (MempoolSnapshot blk)
testSyncWithLedger Mempool IO blk
mempool
    -- Populate mempool (synced against 'tip') with transactions for this slot
    tip <- atomically $ ChainDB.getTipPoint chainDB
    void $
      withEarlyExit $
        ChainDB.withReadOnlyForkerAtPoint chainDB (SpecificPoint tip) $ \case
          Left{} -> do
            IO () -> WithEarlyExit IO ()
forall (m :: * -> *) a. Monad m => m a -> WithEarlyExit m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (IO () -> WithEarlyExit IO ()) -> IO () -> WithEarlyExit IO ()
forall a b. (a -> b) -> a -> b
$ String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"--> could not acquire forker at tip: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Point blk -> String
forall a. Show a => a -> String
show Point blk
tip
            WithEarlyExit IO ()
forall (m :: * -> *) a. Applicative m => WithEarlyExit m a
exitEarly
          Right ReadOnlyForker' IO blk
frk -> do
            extLedgerState <- IO (ExtLedgerState blk EmptyMK)
-> WithEarlyExit IO (ExtLedgerState blk EmptyMK)
forall (m :: * -> *) a. Monad m => m a -> WithEarlyExit m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (IO (ExtLedgerState blk EmptyMK)
 -> WithEarlyExit IO (ExtLedgerState blk EmptyMK))
-> IO (ExtLedgerState blk EmptyMK)
-> WithEarlyExit IO (ExtLedgerState blk EmptyMK)
forall a b. (a -> b) -> a -> b
$ STM IO (ExtLedgerState blk EmptyMK)
-> IO (ExtLedgerState blk EmptyMK)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO (ExtLedgerState blk EmptyMK)
 -> IO (ExtLedgerState blk EmptyMK))
-> STM IO (ExtLedgerState blk EmptyMK)
-> IO (ExtLedgerState blk EmptyMK)
forall a b. (a -> b) -> a -> b
$ ReadOnlyForker' IO blk -> STM IO (ExtLedgerState blk EmptyMK)
forall (m :: * -> *) (l :: StateKind) blk.
ReadOnlyForker m l blk -> STM m (l blk EmptyMK)
roforkerGetLedgerState ReadOnlyForker' IO blk
frk
            let tickedLedgerState =
                  ComputeLedgerEvents
-> LedgerCfg LedgerState blk
-> SlotNo
-> LedgerState blk EmptyMK
-> Ticked LedgerState blk DiffMK
forall (l :: StateKind) blk.
IsLedger l blk =>
ComputeLedgerEvents
-> LedgerCfg l blk
-> SlotNo
-> l blk EmptyMK
-> Ticked l blk DiffMK
applyChainTick
                    ComputeLedgerEvents
OmitLedgerEvents
                    (TopLevelConfig blk -> LedgerCfg LedgerState blk
forall blk. TopLevelConfig blk -> LedgerConfig blk
configLedger TopLevelConfig blk
cfg)
                    SlotNo
currentSlot
                    (ExtLedgerState blk EmptyMK -> LedgerState blk EmptyMK
forall blk (mk :: MapKind).
ExtLedgerState blk mk -> LedgerState blk mk
ledgerState ExtLedgerState blk EmptyMK
extLedgerState)
            txs <- lift $ genTxs currentSlot frk tickedLedgerState
            mayMpRes <- lift $ timeout mempoolAddTxsTimeout $ addLocalTxs mempool txs
            case mayMpRes of
              Maybe [MempoolAddTxResult blk]
Nothing -> IO () -> WithEarlyExit IO ()
forall (m :: * -> *) a. Monad m => m a -> WithEarlyExit m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (IO () -> WithEarlyExit IO ()) -> IO () -> WithEarlyExit IO ()
forall a b. (a -> b) -> a -> b
$ SlotNo -> [GenTx blk] -> IO ()
forall {a} {t :: * -> *} {a}.
(Show a, Foldable t) =>
a -> t a -> IO ()
warnTimeout SlotNo
currentSlot [GenTx blk]
txs
              Just [MempoolAddTxResult blk]
mpRes -> IO () -> WithEarlyExit IO ()
forall (m :: * -> *) a. Monad m => m a -> WithEarlyExit m a
forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift (IO () -> WithEarlyExit IO ()) -> IO () -> WithEarlyExit IO ()
forall a b. (a -> b) -> a -> b
$ SlotNo -> [GenTx blk] -> [MempoolAddTxResult blk] -> IO ()
forall {a} {t :: * -> *} {a} {blk}.
(Show a, Foldable t) =>
a -> t a -> [MempoolAddTxResult blk] -> IO ()
warnRejected SlotNo
currentSlot [GenTx blk]
txs [MempoolAddTxResult blk]
mpRes
    -- Forge a block for each credential.
    -- A block is adopted when forge returns `Just ()`.
    -- If more than one is a leader in this slot, there will be a fork and some block will be orphaned in VolatileDB.
    results <-
      mapM
        (\BlockForging IO blk
bf -> WithEarlyExit IO () -> IO (Maybe ())
forall (m :: * -> *) a. WithEarlyExit m a -> m (Maybe a)
withEarlyExit (WithEarlyExit IO () -> IO (Maybe ()))
-> WithEarlyExit IO () -> IO (Maybe ())
forall a b. (a -> b) -> a -> b
$ Tracer IO (TraceLabelCreds (TraceForgeEvent blk))
-> Tracer IO (TraceLabelCreds (ForgeStateInfo blk))
-> TopLevelConfig blk
-> ChainDB IO blk
-> Mempool IO blk
-> BlockForging IO blk
-> SlotNo
-> WithEarlyExit IO ()
forall (m :: * -> *) blk.
(IOLike m, RunNode blk) =>
Tracer m (TraceLabelCreds (TraceForgeEvent blk))
-> Tracer m (TraceLabelCreds (ForgeStateInfo blk))
-> TopLevelConfig blk
-> ChainDB m blk
-> Mempool m blk
-> BlockForging m blk
-> SlotNo
-> WithEarlyExit m ()
forge Tracer IO (TraceLabelCreds (TraceForgeEvent blk))
forall (m :: * -> *) a. Monad m => Tracer m a
nullTracer Tracer IO (TraceLabelCreds (ForgeStateInfo blk))
forall (m :: * -> *) a. Monad m => Tracer m a
nullTracer TopLevelConfig blk
cfg ChainDB IO blk
chainDB Mempool IO blk
mempool BlockForging IO blk
bf SlotNo
currentSlot)
        blockForging
    pure $ any isJust results

  warnTimeout :: a -> t a -> IO ()
warnTimeout a
slot t a
txs =
    String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$
      String
"--> at slot "
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> a -> String
forall a. Show a => a -> String
show a
slot
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" timeout of"
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int -> String
forall a. Show a => a -> String
show Int
mempoolAddTxsTimeout
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" micros reached when adding "
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> Int -> String
forall a. Show a => a -> String
show (t a -> Int
forall a. t a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length t a
txs)
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" transactions to the mempool"
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"\nMempool is probably full and blocks new additions until it frees up space."
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"\nIt's either that nothing is clearing the Mempool because no new blocks are forged, or"
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"\nthe transaction generator provided too many transactions."
        String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"\nContinuing with forging..."

  warnRejected :: a -> t a -> [MempoolAddTxResult blk] -> IO ()
warnRejected a
slot t a
txs [MempoolAddTxResult blk]
mpRes =
    let rejected :: [ApplyTxErr blk]
rejected = [ApplyTxErr blk
e | MempoolTxRejected GenTx blk
_ ApplyTxErr blk
e <- [MempoolAddTxResult blk]
mpRes]
     in Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([ApplyTxErr blk] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [ApplyTxErr blk]
rejected) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"--> at slot " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> a -> String
forall a. Show a => a -> String
show a
slot String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" generated txs: " String -> String -> String
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show (t a -> Int
forall a. t a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length t a
txs)
          String -> IO ()
putStrLn (String -> IO ()) -> String -> IO ()
forall a b. (a -> b) -> a -> b
$ String
"--> at slot " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> a -> String
forall a. Show a => a -> String
show a
slot String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
" mempool rejected txs: " String -> String -> String
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show ([ApplyTxErr blk] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [ApplyTxErr blk]
rejected)