module Hydra.Events.RotationSpec where

import Hydra.Prelude
import Test.Hydra.Prelude

import Control.Monad (foldM)
import Data.List qualified as List
import Data.Map.Strict qualified as Map
import Hydra.Chain (OnChainTx (..))
import Hydra.Chain.ChainState (IsChainState)
import Hydra.Events (EventId, EventSink (..), HasEventId (..), getEvents)
import Hydra.Events.Rotation (EventStore (..), RotationConfig (..), newRotatedEventStore)
import Hydra.HeadLogic (HeadState (..), StateChanged (..), aggregateNodeState)
import Hydra.HeadLogic.StateEvent (StateEvent (..), mkCheckpoint)
import Hydra.Ledger.Simple (SimpleTx, simpleLedger)
import Hydra.Logging (showLogsOnFailure)
import Hydra.Node (DraftHydraNode, hydrate)
import Hydra.Node.State (NodeState (..), initNodeState)
import Hydra.NodeSpec (createMockEventStore, inputsToOpenHead, notConnect, observationInput, primeWith, primeWithTime, runToCompletion)
import Hydra.Tx.ContestationPeriod (toNominalDiffTime)
import Test.Hydra.Ledger.Simple (utxoRef)
import Test.Hydra.Node.Fixture (testEnvironment, testHeadId)
import Test.Hydra.Tx.Fixture (cperiod)
import Test.QuickCheck (Positive (..), choose, sized)
import Test.QuickCheck.Instances.Natural ()

spec :: Spec
spec :: Spec
spec = Spec -> Spec
forall a. SpecWith a -> SpecWith a
parallel (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
  String -> Spec -> Spec
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Log rotation" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    -- Set up a hydrate function with fixtures curried
    let setupHydrate ::
          ( ( EventStore (StateEvent SimpleTx) IO ->
              [EventSink (StateEvent SimpleTx) IO] ->
              IO (DraftHydraNode SimpleTx IO)
            ) ->
            IO ()
          ) ->
          IO ()
        setupHydrate :: ((EventStore (StateEvent SimpleTx) IO
  -> [EventSink (StateEvent SimpleTx) IO]
  -> IO (DraftHydraNode SimpleTx IO))
 -> IO ())
-> IO ()
setupHydrate (EventStore (StateEvent SimpleTx) IO
 -> [EventSink (StateEvent SimpleTx) IO]
 -> IO (DraftHydraNode SimpleTx IO))
-> IO ()
action =
          Text -> (Tracer IO (HydraNodeLog SimpleTx) -> IO ()) -> IO ()
forall (m :: * -> *) msg a.
(MonadLabelledSTM m, MonadCatch m, MonadFork m, MonadTime m,
 MonadSay m, ToJSON msg) =>
Text -> (Tracer m msg -> m a) -> m a
showLogsOnFailure Text
"RotationSpec" ((Tracer IO (HydraNodeLog SimpleTx) -> IO ()) -> IO ())
-> (Tracer IO (HydraNodeLog SimpleTx) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Tracer IO (HydraNodeLog SimpleTx)
tracer -> do
            let testHydrate :: EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate = Tracer IO (HydraNodeLog SimpleTx)
-> Environment
-> Ledger SimpleTx
-> ChainStateType SimpleTx
-> EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
forall tx (m :: * -> *).
(IsChainState tx, MonadDelay m, MonadLabelledSTM m, MonadAsync m,
 MonadThrow m, MonadUnliftIO m) =>
Tracer m (HydraNodeLog tx)
-> Environment
-> Ledger tx
-> ChainStateType tx
-> EventStore (StateEvent tx) m
-> [EventSink (StateEvent tx) m]
-> m (DraftHydraNode tx m)
hydrate Tracer IO (HydraNodeLog SimpleTx)
tracer Environment
testEnvironment Ledger SimpleTx
simpleLedger ChainStateType SimpleTx
SimpleChainState
0
            (EventStore (StateEvent SimpleTx) IO
 -> [EventSink (StateEvent SimpleTx) IO]
 -> IO (DraftHydraNode SimpleTx IO))
-> IO ()
action EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate
    (((EventStore (StateEvent SimpleTx) IO
   -> [EventSink (StateEvent SimpleTx) IO]
   -> IO (DraftHydraNode SimpleTx IO))
  -> IO ())
 -> IO ())
-> SpecWith
     (EventStore (StateEvent SimpleTx) IO
      -> [EventSink (StateEvent SimpleTx) IO]
      -> IO (DraftHydraNode SimpleTx IO))
-> Spec
forall a. (ActionWith a -> IO ()) -> SpecWith a -> Spec
around ((EventStore (StateEvent SimpleTx) IO
  -> [EventSink (StateEvent SimpleTx) IO]
  -> IO (DraftHydraNode SimpleTx IO))
 -> IO ())
-> IO ()
setupHydrate (SpecWith
   (EventStore (StateEvent SimpleTx) IO
    -> [EventSink (StateEvent SimpleTx) IO]
    -> IO (DraftHydraNode SimpleTx IO))
 -> Spec)
-> SpecWith
     (EventStore (StateEvent SimpleTx) IO
      -> [EventSink (StateEvent SimpleTx) IO]
      -> IO (DraftHydraNode SimpleTx IO))
-> Spec
forall a b. (a -> b) -> a -> b
$ do
      String
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"rotates while running" (((EventStore (StateEvent SimpleTx) IO
   -> [EventSink (StateEvent SimpleTx) IO]
   -> IO (DraftHydraNode SimpleTx IO))
  -> IO ())
 -> SpecWith
      (Arg
         ((EventStore (StateEvent SimpleTx) IO
           -> [EventSink (StateEvent SimpleTx) IO]
           -> IO (DraftHydraNode SimpleTx IO))
          -> IO ())))
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
10 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          EventStore (StateEvent SimpleTx) IO
eventStore <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          -- NOTE: because there will be 5 inputs processed in total, after ticks,
          -- this is hardcoded to ensure we get a checkpoint + single event at the end
          let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
2)
          let s0 :: NodeState SimpleTx
s0 = ChainStateType SimpleTx -> NodeState SimpleTx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType SimpleTx
0
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
eventStore
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *).
MonadSTM m =>
HydraNode SimpleTx m -> m (HydraNode SimpleTx m)
primeWithTime
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputsToOpenHead
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          [StateEvent SimpleTx]
rotatedHistory <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
rotatingEventStore)
          [StateEvent SimpleTx] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [StateEvent SimpleTx]
rotatedHistory Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Int
1
      String
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"consistent state after restarting with rotation" (((EventStore (StateEvent SimpleTx) IO
   -> [EventSink (StateEvent SimpleTx) IO]
   -> IO (DraftHydraNode SimpleTx IO))
  -> IO ())
 -> SpecWith
      (Arg
         ((EventStore (StateEvent SimpleTx) IO
           -> [EventSink (StateEvent SimpleTx) IO]
           -> IO (DraftHydraNode SimpleTx IO))
          -> IO ())))
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
10 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          EventStore (StateEvent SimpleTx) IO
eventStore <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          -- NOTE: because there will be 6 inputs processed in total, after ticks,
          -- this is hardcoded to ensure we get a single checkpoint event at the end
          let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
1)
          let s0 :: NodeState SimpleTx
s0 = ChainStateType SimpleTx -> NodeState SimpleTx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType SimpleTx
0
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
eventStore
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *).
MonadSTM m =>
HydraNode SimpleTx m -> m (HydraNode SimpleTx m)
primeWithTime
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputsToOpenHead
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          UTCTime
now <- IO UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
          let contestationDeadline :: UTCTime
contestationDeadline = ContestationPeriod -> NominalDiffTime
toNominalDiffTime ContestationPeriod
cperiod NominalDiffTime -> UTCTime -> UTCTime
`addUTCTime` UTCTime
now
          let closeInput :: Input SimpleTx
closeInput = OnChainTx SimpleTx -> Input SimpleTx
observationInput (OnChainTx SimpleTx -> Input SimpleTx)
-> OnChainTx SimpleTx -> Input SimpleTx
forall a b. (a -> b) -> a -> b
$ HeadId -> SnapshotNumber -> UTCTime -> OnChainTx SimpleTx
forall tx. HeadId -> SnapshotNumber -> UTCTime -> OnChainTx tx
OnCloseTx HeadId
testHeadId SnapshotNumber
0 UTCTime
contestationDeadline
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx
closeInput]
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          [StateEvent SimpleTx
checkpoint] <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
rotatingEventStore)
          case StateEvent SimpleTx -> StateChanged SimpleTx
forall tx. StateEvent tx -> StateChanged tx
stateChanged StateEvent SimpleTx
checkpoint of
            Checkpoint{$sel:state:NetworkConnected :: forall tx. StateChanged tx -> NodeState tx
state = NodeInSync{$sel:headState:NodeInSync :: forall tx. NodeState tx -> HeadState tx
headState = Closed{}}} -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
            StateChanged SimpleTx
_ -> String -> IO ()
forall a. String -> IO a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail (String
"unexpected: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> StateEvent SimpleTx -> String
forall b a. (Show a, IsString b) => a -> b
show StateEvent SimpleTx
checkpoint)
      String
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"preserves pending deposits after restarting with rotation" (((EventStore (StateEvent SimpleTx) IO
   -> [EventSink (StateEvent SimpleTx) IO]
   -> IO (DraftHydraNode SimpleTx IO))
  -> IO ())
 -> SpecWith
      (Arg
         ((EventStore (StateEvent SimpleTx) IO
           -> [EventSink (StateEvent SimpleTx) IO]
           -> IO (DraftHydraNode SimpleTx IO))
          -> IO ())))
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate -> do
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
10 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          EventStore (StateEvent SimpleTx) IO
eventStore <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          -- Rotate aggressively so the stored history ends with a single
          -- checkpoint capturing the recorded deposit.
          let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
1)
          let s0 :: NodeState SimpleTx
s0 = ChainStateType SimpleTx -> NodeState SimpleTx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType SimpleTx
0
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
eventStore
          UTCTime
now <- IO UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
          let deadline :: UTCTime
deadline = ContestationPeriod -> NominalDiffTime
toNominalDiffTime ContestationPeriod
cperiod NominalDiffTime -> UTCTime -> UTCTime
`addUTCTime` UTCTime
now
          let depositTxId :: SimpleId
depositTxId = SimpleId
1
          let depositInput :: Input SimpleTx
depositInput = OnChainTx SimpleTx -> Input SimpleTx
observationInput (OnChainTx SimpleTx -> Input SimpleTx)
-> OnChainTx SimpleTx -> Input SimpleTx
forall a b. (a -> b) -> a -> b
$ HeadId
-> TxIdType SimpleTx
-> UTxOType SimpleTx
-> UTCTime
-> UTCTime
-> OnChainTx SimpleTx
forall tx.
HeadId
-> TxIdType tx -> UTxOType tx -> UTCTime -> UTCTime -> OnChainTx tx
OnDepositTx HeadId
testHeadId SimpleId
TxIdType SimpleTx
depositTxId (SimpleId -> UTxOType SimpleTx
utxoRef SimpleId
1) UTCTime
now UTCTime
deadline
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *).
MonadSTM m =>
HydraNode SimpleTx m -> m (HydraNode SimpleTx m)
primeWithTime
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith ([Input SimpleTx]
inputsToOpenHead [Input SimpleTx] -> [Input SimpleTx] -> [Input SimpleTx]
forall a. Semigroup a => a -> a -> a
<> [Input SimpleTx
depositInput])
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          -- Restart reconstructs the node state exactly like 'hydrate' does:
          -- fold the stored events (a single checkpoint) with 'aggregateNodeState'.
          [StateEvent SimpleTx]
events <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
rotatingEventStore)
          let restored :: NodeState SimpleTx
restored = (NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx)
-> NodeState SimpleTx
-> [StateEvent SimpleTx]
-> NodeState SimpleTx
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx
s0 [StateEvent SimpleTx]
events
          Map SimpleId (Deposit SimpleTx) -> [SimpleId]
forall k a. Map k a -> [k]
Map.keys (NodeState SimpleTx -> PendingDeposits SimpleTx
forall tx. NodeState tx -> PendingDeposits tx
pendingDeposits NodeState SimpleTx
restored) [SimpleId] -> [SimpleId] -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` [SimpleId
depositTxId]
      String
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"a rotated and non-rotated node have consistent state" (((EventStore (StateEvent SimpleTx) IO
   -> [EventSink (StateEvent SimpleTx) IO]
   -> IO (DraftHydraNode SimpleTx IO))
  -> IO ())
 -> SpecWith
      (Arg
         ((EventStore (StateEvent SimpleTx) IO
           -> [EventSink (StateEvent SimpleTx) IO]
           -> IO (DraftHydraNode SimpleTx IO))
          -> IO ())))
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate -> do
        -- prepare inputs
        UTCTime
now <- IO UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
        let contestationDeadline :: UTCTime
contestationDeadline = ContestationPeriod -> NominalDiffTime
toNominalDiffTime ContestationPeriod
cperiod NominalDiffTime -> UTCTime -> UTCTime
`addUTCTime` UTCTime
now
        let closeInput :: Input SimpleTx
closeInput = OnChainTx SimpleTx -> Input SimpleTx
observationInput (OnChainTx SimpleTx -> Input SimpleTx)
-> OnChainTx SimpleTx -> Input SimpleTx
forall a b. (a -> b) -> a -> b
$ HeadId -> SnapshotNumber -> UTCTime -> OnChainTx SimpleTx
forall tx. HeadId -> SnapshotNumber -> UTCTime -> OnChainTx tx
OnCloseTx HeadId
testHeadId SnapshotNumber
0 UTCTime
contestationDeadline
        let inputs :: [Input SimpleTx]
inputs = [Input SimpleTx]
inputsToOpenHead [Input SimpleTx] -> [Input SimpleTx] -> [Input SimpleTx]
forall a. [a] -> [a] -> [a]
++ [Input SimpleTx
closeInput]
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
10 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          EventStore (StateEvent SimpleTx) IO
eventStore <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          -- NOTE: because there will be 6 inputs processed in total, after ticks,
          -- this is hardcoded to ensure we get a single checkpoint event at the end
          let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
1)
          -- run rotated event store with prepared inputs
          let s0 :: NodeState SimpleTx
s0 = ChainStateType SimpleTx -> NodeState SimpleTx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType SimpleTx
0
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
eventStore
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputs
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          -- run non-rotated event store with prepared inputs
          EventStore (StateEvent SimpleTx) IO
eventStore' <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
eventStore' []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputs
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          -- aggregating stored events should yield consistent states
          [StateEvent{$sel:stateChanged:StateEvent :: forall tx. StateEvent tx -> StateChanged tx
stateChanged = StateChanged SimpleTx
checkpoint}] <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
rotatingEventStore)
          [StateEvent SimpleTx]
events' <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
eventStore')
          let checkpoint' :: NodeState SimpleTx
checkpoint' = (NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx)
-> NodeState SimpleTx
-> [StateEvent SimpleTx]
-> NodeState SimpleTx
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx
s0 [StateEvent SimpleTx]
events'
          StateChanged SimpleTx
checkpoint StateChanged SimpleTx -> StateChanged SimpleTx -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` NodeState SimpleTx -> StateChanged SimpleTx
forall tx. NodeState tx -> StateChanged tx
Checkpoint NodeState SimpleTx
checkpoint'
      String
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"a restarted and non-restarted node have consistent rotation" (((EventStore (StateEvent SimpleTx) IO
   -> [EventSink (StateEvent SimpleTx) IO]
   -> IO (DraftHydraNode SimpleTx IO))
  -> IO ())
 -> SpecWith
      (Arg
         ((EventStore (StateEvent SimpleTx) IO
           -> [EventSink (StateEvent SimpleTx) IO]
           -> IO (DraftHydraNode SimpleTx IO))
          -> IO ())))
-> ((EventStore (StateEvent SimpleTx) IO
     -> [EventSink (StateEvent SimpleTx) IO]
     -> IO (DraftHydraNode SimpleTx IO))
    -> IO ())
-> SpecWith
     (Arg
        ((EventStore (StateEvent SimpleTx) IO
          -> [EventSink (StateEvent SimpleTx) IO]
          -> IO (DraftHydraNode SimpleTx IO))
         -> IO ()))
forall a b. (a -> b) -> a -> b
$ \EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate -> do
        -- prepare inputs
        UTCTime
now <- IO UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
        let contestationDeadline :: UTCTime
contestationDeadline = ContestationPeriod -> NominalDiffTime
toNominalDiffTime ContestationPeriod
cperiod NominalDiffTime -> UTCTime -> UTCTime
`addUTCTime` UTCTime
now
        let closeInput :: Input SimpleTx
closeInput = OnChainTx SimpleTx -> Input SimpleTx
observationInput (OnChainTx SimpleTx -> Input SimpleTx)
-> OnChainTx SimpleTx -> Input SimpleTx
forall a b. (a -> b) -> a -> b
$ HeadId -> SnapshotNumber -> UTCTime -> OnChainTx SimpleTx
forall tx. HeadId -> SnapshotNumber -> UTCTime -> OnChainTx tx
OnCloseTx HeadId
testHeadId SnapshotNumber
0 UTCTime
contestationDeadline
        let inputs :: [Input SimpleTx]
inputs = [Input SimpleTx]
inputsToOpenHead [Input SimpleTx] -> [Input SimpleTx] -> [Input SimpleTx]
forall a. [a] -> [a] -> [a]
++ [Input SimpleTx
closeInput]
        let inputs1 :: [Input SimpleTx]
inputs1 = Int -> [Input SimpleTx] -> [Input SimpleTx]
forall a. Int -> [a] -> [a]
take Int
3 [Input SimpleTx]
inputs
        let inputs2 :: [Input SimpleTx]
inputs2 = Int -> [Input SimpleTx] -> [Input SimpleTx]
forall a. Int -> [a] -> [a]
drop Int
3 [Input SimpleTx]
inputs
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
10 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          let s0 :: NodeState SimpleTx
s0 = ChainStateType SimpleTx -> NodeState SimpleTx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType SimpleTx
0
          -- NOTE: because there will be 6 inputs processed in total, after ticks,
          -- this is hardcoded to ensure we get a single checkpoint event at the end
          let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
1)
          -- run restarted node with prepared inputs
          EventStore (StateEvent SimpleTx) IO
eventStore <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore1 <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
eventStore
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore1 []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *).
MonadSTM m =>
HydraNode SimpleTx m -> m (HydraNode SimpleTx m)
primeWithTime
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputs1
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore2 <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
rotatingEventStore1
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore2 []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputs2
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          -- run non-restarted node with prepared inputs
          EventStore (StateEvent SimpleTx) IO
eventStore' <- IO (EventStore (StateEvent SimpleTx) IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          EventStore (StateEvent SimpleTx) IO
rotatingEventStore' <- RotationConfig
-> NodeState SimpleTx
-> (NodeState SimpleTx
    -> StateEvent SimpleTx -> NodeState SimpleTx)
-> (NodeState SimpleTx
    -> EventId -> UTCTime -> StateEvent SimpleTx)
-> EventStore (StateEvent SimpleTx) IO
-> IO (EventStore (StateEvent SimpleTx) IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig NodeState SimpleTx
s0 NodeState SimpleTx -> StateEvent SimpleTx -> NodeState SimpleTx
forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState SimpleTx -> EventId -> UTCTime -> StateEvent SimpleTx
forall tx. NodeState tx -> EventId -> UTCTime -> StateEvent tx
mkCheckpoint EventStore (StateEvent SimpleTx) IO
eventStore'
          EventStore (StateEvent SimpleTx) IO
-> [EventSink (StateEvent SimpleTx) IO]
-> IO (DraftHydraNode SimpleTx IO)
testHydrate EventStore (StateEvent SimpleTx) IO
rotatingEventStore' []
            IO (DraftHydraNode SimpleTx IO)
-> (DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= DraftHydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadThrow m =>
DraftHydraNode tx m -> m (HydraNode tx m)
notConnect
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *).
MonadSTM m =>
HydraNode SimpleTx m -> m (HydraNode SimpleTx m)
primeWithTime
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO))
-> IO (HydraNode SimpleTx IO)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [Input SimpleTx]
-> HydraNode SimpleTx IO -> IO (HydraNode SimpleTx IO)
forall (m :: * -> *) tx.
MonadSTM m =>
[Input tx] -> HydraNode tx m -> m (HydraNode tx m)
primeWith [Input SimpleTx]
inputs
            IO (HydraNode SimpleTx IO)
-> (HydraNode SimpleTx IO -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= HydraNode SimpleTx IO -> IO ()
forall tx. IsChainState tx => HydraNode tx IO -> IO ()
runToCompletion
          -- stored events should yield consistent checkpoint events
          [StateEvent{$sel:eventId:StateEvent :: forall tx. StateEvent tx -> EventId
eventId = EventId
eventId, $sel:stateChanged:StateEvent :: forall tx. StateEvent tx -> StateChanged tx
stateChanged = StateChanged SimpleTx
checkpoint}] <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
rotatingEventStore2)
          [StateEvent{$sel:eventId:StateEvent :: forall tx. StateEvent tx -> EventId
eventId = EventId
eventId', $sel:stateChanged:StateEvent :: forall tx. StateEvent tx -> StateChanged tx
stateChanged = StateChanged SimpleTx
checkpoint'}] <- EventSource (StateEvent SimpleTx) IO -> IO [StateEvent SimpleTx]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents (EventStore (StateEvent SimpleTx) IO
-> EventSource (StateEvent SimpleTx) IO
forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource EventStore (StateEvent SimpleTx) IO
rotatingEventStore')
          StateChanged SimpleTx
checkpoint StateChanged SimpleTx -> StateChanged SimpleTx -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` StateChanged SimpleTx
checkpoint'
          -- stored events should yield consistent event id
          EventId
eventId EventId -> EventId -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` EventId
eventId'

  String -> Spec -> Spec
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"Rotation algorithm" (Spec -> Spec) -> Spec -> Spec
forall a b. (a -> b) -> a -> b
$ do
    String -> ((Positive Natural, Positive Natural) -> IO ()) -> Spec
forall prop.
(HasCallStack, Testable prop) =>
String -> prop -> Spec
prop String
"rotates on startup" (((Positive Natural, Positive Natural) -> IO ()) -> Spec)
-> ((Positive Natural, Positive Natural) -> IO ()) -> Spec
forall a b. (a -> b) -> a -> b
$
      \(Positive Natural
x, Positive Natural
delta) -> do
        eventStore :: EventStore TrivialEvent IO
eventStore@EventStore{EventSource TrivialEvent IO
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource :: EventSource TrivialEvent IO
eventSource, EventSink TrivialEvent IO
eventSink :: EventSink TrivialEvent IO
$sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink} <- IO (EventStore TrivialEvent IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
        let y :: Natural
y = Natural
x Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
delta
        let totalEvents :: SimpleId
totalEvents = Natural -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger Natural
y
        let events :: [TrivialEvent]
events = EventId -> TrivialEvent
TrivialEvent (EventId -> TrivialEvent) -> [EventId] -> [TrivialEvent]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [EventId
1 .. SimpleId -> EventId
forall a. Num a => SimpleId -> a
fromInteger SimpleId
totalEvents]
        (TrivialEvent -> IO ()) -> [TrivialEvent] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (EventSink TrivialEvent IO
-> HasEventId TrivialEvent => TrivialEvent -> IO ()
forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent EventSink TrivialEvent IO
eventSink) [TrivialEvent]
events
        [TrivialEvent]
unrotatedHistory <- EventSource TrivialEvent IO -> IO [TrivialEvent]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource TrivialEvent IO
eventSource
        Int -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger ([TrivialEvent] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [TrivialEvent]
unrotatedHistory) SimpleId -> SimpleId -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` SimpleId
totalEvents
        let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
x)
        let s0 :: [TrivialEvent]
            s0 :: [TrivialEvent]
s0 = []
        let aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
            aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent]
s TrivialEvent
e = TrivialEvent
e TrivialEvent -> [TrivialEvent] -> [TrivialEvent]
forall a. a -> [a] -> [a]
: [TrivialEvent]
s
        let checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
            checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer [TrivialEvent]
s EventId
_ UTCTime
_ = [TrivialEvent] -> TrivialEvent
trivialCheckpoint [TrivialEvent]
s
        EventStore{$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource = EventSource TrivialEvent IO
rotatedEventSource} <- RotationConfig
-> [TrivialEvent]
-> ([TrivialEvent] -> TrivialEvent -> [TrivialEvent])
-> ([TrivialEvent] -> EventId -> UTCTime -> TrivialEvent)
-> EventStore TrivialEvent IO
-> IO (EventStore TrivialEvent IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig [TrivialEvent]
s0 [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer EventStore TrivialEvent IO
eventStore
        [TrivialEvent]
rotatedHistory <- EventSource TrivialEvent IO -> IO [TrivialEvent]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource TrivialEvent IO
rotatedEventSource
        [TrivialEvent] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [TrivialEvent]
rotatedHistory Int -> Int -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` Int
1

    -- given some event store (source + sink)
    -- lets configure a rotated event store that rotates after x events
    -- forall y > 0: put x*y+1 events
    -- load all events returns a suffix of put events with length <= x
    String -> ((Positive Natural, Positive SimpleId) -> IO ()) -> Spec
forall prop.
(HasCallStack, Testable prop) =>
String -> prop -> Spec
prop String
"rotates after configured number of events" (((Positive Natural, Positive SimpleId) -> IO ()) -> Spec)
-> ((Positive Natural, Positive SimpleId) -> IO ()) -> Spec
forall a b. (a -> b) -> a -> b
$
      \(Positive Natural
x, Positive SimpleId
y) -> do
        EventStore TrivialEvent IO
mockEventStore <- IO (EventStore TrivialEvent IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
        let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
x)
        let s0 :: [TrivialEvent]
            s0 :: [TrivialEvent]
s0 = []
        let aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
            aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent]
s TrivialEvent
e = TrivialEvent
e TrivialEvent -> [TrivialEvent] -> [TrivialEvent]
forall a. a -> [a] -> [a]
: [TrivialEvent]
s
        let checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
            checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer [TrivialEvent]
s EventId
_ UTCTime
_ = [TrivialEvent] -> TrivialEvent
trivialCheckpoint [TrivialEvent]
s
        EventStore TrivialEvent IO
rotatingEventStore <- RotationConfig
-> [TrivialEvent]
-> ([TrivialEvent] -> TrivialEvent -> [TrivialEvent])
-> ([TrivialEvent] -> EventId -> UTCTime -> TrivialEvent)
-> EventStore TrivialEvent IO
-> IO (EventStore TrivialEvent IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig [TrivialEvent]
s0 [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer EventStore TrivialEvent IO
mockEventStore
        let EventStore{EventSource TrivialEvent IO
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource :: EventSource TrivialEvent IO
eventSource, $sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink = EventSink{HasEventId TrivialEvent => TrivialEvent -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId TrivialEvent => TrivialEvent -> IO ()
putEvent}} = EventStore TrivialEvent IO
rotatingEventStore
        let totalEvents :: SimpleId
totalEvents = Natural -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger Natural
x SimpleId -> SimpleId -> SimpleId
forall a. Num a => a -> a -> a
* SimpleId
y SimpleId -> SimpleId -> SimpleId
forall a. Num a => a -> a -> a
+ SimpleId
1
        let events :: [TrivialEvent]
events = EventId -> TrivialEvent
TrivialEvent (EventId -> TrivialEvent)
-> (SimpleId -> EventId) -> SimpleId -> TrivialEvent
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SimpleId -> EventId
forall a. Num a => SimpleId -> a
fromInteger (SimpleId -> TrivialEvent) -> [SimpleId] -> [TrivialEvent]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [SimpleId
1 .. SimpleId
totalEvents]
        [TrivialEvent] -> (TrivialEvent -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TrivialEvent]
events HasEventId TrivialEvent => TrivialEvent -> IO ()
TrivialEvent -> IO ()
putEvent
        [TrivialEvent]
currentHistory <- EventSource TrivialEvent IO -> IO [TrivialEvent]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource TrivialEvent IO
eventSource
        let expectRotated :: [TrivialEvent]
expectRotated = [TrivialEvent]
events
        let [TrivialEvent]
expectRemaining :: [TrivialEvent] = []
        let expectedCurrentHistory :: [TrivialEvent]
expectedCurrentHistory = [TrivialEvent] -> TrivialEvent
trivialCheckpoint [TrivialEvent]
expectRotated TrivialEvent -> [TrivialEvent] -> [TrivialEvent]
forall a. a -> [a] -> [a]
: [TrivialEvent]
expectRemaining
        [TrivialEvent]
expectedCurrentHistory [TrivialEvent] -> [TrivialEvent] -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` [TrivialEvent]
currentHistory

    -- forall y. y > 0 && y < x: put x+y events (= ensures rotation)
    -- checkpoint of first x + 1 of events === load first event
    String -> ((Positive Natural, Positive Natural) -> IO ()) -> Spec
forall prop.
(HasCallStack, Testable prop) =>
String -> prop -> Spec
prop String
"puts checkpoint event as first event" (((Positive Natural, Positive Natural) -> IO ()) -> Spec)
-> ((Positive Natural, Positive Natural) -> IO ()) -> Spec
forall a b. (a -> b) -> a -> b
$
      \(Positive Natural
y, Positive Natural
delta) -> do
        let x :: Natural
x = Natural
y Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
delta
        EventStore TrivialEvent IO
mockEventStore <- IO (EventStore TrivialEvent IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
        let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter (Natural -> Positive Natural
forall a. a -> Positive a
Positive Natural
x)
        let s0 :: [TrivialEvent]
            s0 :: [TrivialEvent]
s0 = []
        let aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
            aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent]
s TrivialEvent
e = TrivialEvent
e TrivialEvent -> [TrivialEvent] -> [TrivialEvent]
forall a. a -> [a] -> [a]
: [TrivialEvent]
s
        let checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
            checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer [TrivialEvent]
s EventId
_ UTCTime
_ = [TrivialEvent] -> TrivialEvent
trivialCheckpoint [TrivialEvent]
s
        EventStore TrivialEvent IO
rotatingEventStore <- RotationConfig
-> [TrivialEvent]
-> ([TrivialEvent] -> TrivialEvent -> [TrivialEvent])
-> ([TrivialEvent] -> EventId -> UTCTime -> TrivialEvent)
-> EventStore TrivialEvent IO
-> IO (EventStore TrivialEvent IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig [TrivialEvent]
s0 [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer EventStore TrivialEvent IO
mockEventStore
        let EventStore{EventSource TrivialEvent IO
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource :: EventSource TrivialEvent IO
eventSource, $sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink = EventSink{HasEventId TrivialEvent => TrivialEvent -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId TrivialEvent => TrivialEvent -> IO ()
putEvent}} = EventStore TrivialEvent IO
rotatingEventStore
        let totalEvents :: SimpleId
totalEvents = Natural -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger Natural
x SimpleId -> SimpleId -> SimpleId
forall a. Num a => a -> a -> a
+ Natural -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger Natural
y
        let events :: [TrivialEvent]
events = EventId -> TrivialEvent
TrivialEvent (EventId -> TrivialEvent)
-> (SimpleId -> EventId) -> SimpleId -> TrivialEvent
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SimpleId -> EventId
forall a. Num a => SimpleId -> a
fromInteger (SimpleId -> TrivialEvent) -> [SimpleId] -> [TrivialEvent]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [SimpleId
1 .. SimpleId
totalEvents]
        [TrivialEvent] -> (TrivialEvent -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TrivialEvent]
events HasEventId TrivialEvent => TrivialEvent -> IO ()
TrivialEvent -> IO ()
putEvent
        [TrivialEvent]
currentHistory <- EventSource TrivialEvent IO -> IO [TrivialEvent]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource TrivialEvent IO
eventSource
        let expectRotated :: [TrivialEvent]
expectRotated = Int -> [TrivialEvent] -> [TrivialEvent]
forall a. Int -> [a] -> [a]
take (SimpleId -> Int
forall a. Num a => SimpleId -> a
fromInteger (SimpleId -> Int) -> SimpleId -> Int
forall a b. (a -> b) -> a -> b
$ Natural -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger Natural
x SimpleId -> SimpleId -> SimpleId
forall a. Num a => a -> a -> a
+ SimpleId
1) [TrivialEvent]
events
        [TrivialEvent] -> TrivialEvent
trivialCheckpoint [TrivialEvent]
expectRotated TrivialEvent -> TrivialEvent -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` [TrivialEvent] -> TrivialEvent
forall a. HasCallStack => [a] -> a
List.head [TrivialEvent]
currentHistory

    String -> ((Positive Natural, ChunkedEvents) -> IO ()) -> Spec
forall prop.
(HasCallStack, Testable prop) =>
String -> prop -> Spec
prop String
"a restarted and non-restarted store have consistent rotation" (((Positive Natural, ChunkedEvents) -> IO ()) -> Spec)
-> ((Positive Natural, ChunkedEvents) -> IO ()) -> Spec
forall a b. (a -> b) -> a -> b
$
      \(Positive Natural
x, ChunkedEvents [[TrivialEvent]]
chunks) -> do
        let rotationConfig :: RotationConfig
rotationConfig = Positive Natural -> RotationConfig
RotateAfter Positive Natural
x
        let s0 :: [TrivialEvent]
            s0 :: [TrivialEvent]
s0 = []
        let aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
            aggregator :: [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent]
s TrivialEvent
e = TrivialEvent
e TrivialEvent -> [TrivialEvent] -> [TrivialEvent]
forall a. a -> [a] -> [a]
: [TrivialEvent]
s
        let checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
            checkpointer :: [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer [TrivialEvent]
s EventId
_ UTCTime
_ = [TrivialEvent] -> TrivialEvent
trivialCheckpoint [TrivialEvent]
s
        NominalDiffTime -> IO () -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadTimer m, MonadThrow m) =>
NominalDiffTime -> m a -> m a
failAfter NominalDiffTime
10 (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          -- run restarted in chunks
          EventStore TrivialEvent IO
mockEventStore <- IO (EventStore TrivialEvent IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          EventStore{$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource = EventSource TrivialEvent IO
restartedEventSource} <-
            (EventStore TrivialEvent IO
 -> [TrivialEvent] -> IO (EventStore TrivialEvent IO))
-> EventStore TrivialEvent IO
-> [[TrivialEvent]]
-> IO (EventStore TrivialEvent IO)
forall (t :: * -> *) (m :: * -> *) b a.
(Foldable t, Monad m) =>
(b -> a -> m b) -> b -> t a -> m b
foldM
              ( \EventStore TrivialEvent IO
store [TrivialEvent]
chunk -> do
                  EventStore TrivialEvent IO
rotated <- RotationConfig
-> [TrivialEvent]
-> ([TrivialEvent] -> TrivialEvent -> [TrivialEvent])
-> ([TrivialEvent] -> EventId -> UTCTime -> TrivialEvent)
-> EventStore TrivialEvent IO
-> IO (EventStore TrivialEvent IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig [TrivialEvent]
s0 [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer EventStore TrivialEvent IO
store
                  let EventStore{$sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink = EventSink{HasEventId TrivialEvent => TrivialEvent -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId TrivialEvent => TrivialEvent -> IO ()
putEvent}} = EventStore TrivialEvent IO
rotated
                  [TrivialEvent] -> (TrivialEvent -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TrivialEvent]
chunk HasEventId TrivialEvent => TrivialEvent -> IO ()
TrivialEvent -> IO ()
putEvent
                  EventStore TrivialEvent IO -> IO (EventStore TrivialEvent IO)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure EventStore TrivialEvent IO
rotated
              )
              EventStore TrivialEvent IO
mockEventStore
              [[TrivialEvent]]
chunks
          -- run non-restarted full
          EventStore TrivialEvent IO
mockEventStore' <- IO (EventStore TrivialEvent IO)
forall (m :: * -> *) a. MonadLabelledSTM m => m (EventStore a m)
createMockEventStore
          EventStore TrivialEvent IO
rotated' <- RotationConfig
-> [TrivialEvent]
-> ([TrivialEvent] -> TrivialEvent -> [TrivialEvent])
-> ([TrivialEvent] -> EventId -> UTCTime -> TrivialEvent)
-> EventStore TrivialEvent IO
-> IO (EventStore TrivialEvent IO)
forall e (m :: * -> *) s.
(HasEventId e, MonadUnliftIO m, MonadTime m, MonadLabelledSTM m) =>
RotationConfig
-> s
-> (s -> e -> s)
-> (s -> EventId -> UTCTime -> e)
-> EventStore e m
-> m (EventStore e m)
newRotatedEventStore RotationConfig
rotationConfig [TrivialEvent]
s0 [TrivialEvent] -> TrivialEvent -> [TrivialEvent]
aggregator [TrivialEvent] -> EventId -> UTCTime -> TrivialEvent
checkpointer EventStore TrivialEvent IO
mockEventStore'
          let EventStore{$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource = EventSource TrivialEvent IO
nonRestartedEventSource, $sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink = EventSink{HasEventId TrivialEvent => TrivialEvent -> IO ()
$sel:putEvent:EventSink :: forall e (m :: * -> *). EventSink e m -> HasEventId e => e -> m ()
putEvent :: HasEventId TrivialEvent => TrivialEvent -> IO ()
putEvent}} = EventStore TrivialEvent IO
rotated'
          let events :: [TrivialEvent]
events = [[TrivialEvent]] -> [TrivialEvent]
forall (t :: * -> *) a. Foldable t => t [a] -> [a]
concat [[TrivialEvent]]
chunks
          [TrivialEvent] -> (TrivialEvent -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TrivialEvent]
events HasEventId TrivialEvent => TrivialEvent -> IO ()
TrivialEvent -> IO ()
putEvent
          -- stored events should yield consistent checkpoint event ids
          [TrivialEvent]
restartedHistory <- EventSource TrivialEvent IO -> IO [TrivialEvent]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource TrivialEvent IO
restartedEventSource
          [TrivialEvent]
nonRestartedHistory <- EventSource TrivialEvent IO -> IO [TrivialEvent]
forall e (m :: * -> *).
(HasEventId e, MonadUnliftIO m) =>
EventSource e m -> m [e]
getEvents EventSource TrivialEvent IO
nonRestartedEventSource
          TrivialEvent -> EventId
forall a. HasEventId a => a -> EventId
getEventId ([TrivialEvent] -> TrivialEvent
forall a. HasCallStack => [a] -> a
List.last [TrivialEvent]
restartedHistory) EventId -> EventId -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` TrivialEvent -> EventId
forall a. HasEventId a => a -> EventId
getEventId ([TrivialEvent] -> TrivialEvent
forall a. HasCallStack => [a] -> a
List.last [TrivialEvent]
nonRestartedHistory)

newtype TrivialEvent = TrivialEvent Word64
  deriving newtype (SimpleId -> TrivialEvent
TrivialEvent -> TrivialEvent
TrivialEvent -> TrivialEvent -> TrivialEvent
(TrivialEvent -> TrivialEvent -> TrivialEvent)
-> (TrivialEvent -> TrivialEvent -> TrivialEvent)
-> (TrivialEvent -> TrivialEvent -> TrivialEvent)
-> (TrivialEvent -> TrivialEvent)
-> (TrivialEvent -> TrivialEvent)
-> (TrivialEvent -> TrivialEvent)
-> (SimpleId -> TrivialEvent)
-> Num TrivialEvent
forall a.
(a -> a -> a)
-> (a -> a -> a)
-> (a -> a -> a)
-> (a -> a)
-> (a -> a)
-> (a -> a)
-> (SimpleId -> a)
-> Num a
$c+ :: TrivialEvent -> TrivialEvent -> TrivialEvent
+ :: TrivialEvent -> TrivialEvent -> TrivialEvent
$c- :: TrivialEvent -> TrivialEvent -> TrivialEvent
- :: TrivialEvent -> TrivialEvent -> TrivialEvent
$c* :: TrivialEvent -> TrivialEvent -> TrivialEvent
* :: TrivialEvent -> TrivialEvent -> TrivialEvent
$cnegate :: TrivialEvent -> TrivialEvent
negate :: TrivialEvent -> TrivialEvent
$cabs :: TrivialEvent -> TrivialEvent
abs :: TrivialEvent -> TrivialEvent
$csignum :: TrivialEvent -> TrivialEvent
signum :: TrivialEvent -> TrivialEvent
$cfromInteger :: SimpleId -> TrivialEvent
fromInteger :: SimpleId -> TrivialEvent
Num, Int -> TrivialEvent -> String -> String
[TrivialEvent] -> String -> String
TrivialEvent -> String
(Int -> TrivialEvent -> String -> String)
-> (TrivialEvent -> String)
-> ([TrivialEvent] -> String -> String)
-> Show TrivialEvent
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> TrivialEvent -> String -> String
showsPrec :: Int -> TrivialEvent -> String -> String
$cshow :: TrivialEvent -> String
show :: TrivialEvent -> String
$cshowList :: [TrivialEvent] -> String -> String
showList :: [TrivialEvent] -> String -> String
Show, TrivialEvent -> TrivialEvent -> Bool
(TrivialEvent -> TrivialEvent -> Bool)
-> (TrivialEvent -> TrivialEvent -> Bool) -> Eq TrivialEvent
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: TrivialEvent -> TrivialEvent -> Bool
== :: TrivialEvent -> TrivialEvent -> Bool
$c/= :: TrivialEvent -> TrivialEvent -> Bool
/= :: TrivialEvent -> TrivialEvent -> Bool
Eq)

newtype ChunkedEvents = ChunkedEvents [[TrivialEvent]]
  deriving stock (Int -> ChunkedEvents -> String -> String
[ChunkedEvents] -> String -> String
ChunkedEvents -> String
(Int -> ChunkedEvents -> String -> String)
-> (ChunkedEvents -> String)
-> ([ChunkedEvents] -> String -> String)
-> Show ChunkedEvents
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> ChunkedEvents -> String -> String
showsPrec :: Int -> ChunkedEvents -> String -> String
$cshow :: ChunkedEvents -> String
show :: ChunkedEvents -> String
$cshowList :: [ChunkedEvents] -> String -> String
showList :: [ChunkedEvents] -> String -> String
Show)

instance Arbitrary ChunkedEvents where
  arbitrary :: Gen ChunkedEvents
arbitrary = (Int -> Gen ChunkedEvents) -> Gen ChunkedEvents
forall a. (Int -> Gen a) -> Gen a
sized ((Int -> Gen ChunkedEvents) -> Gen ChunkedEvents)
-> (Int -> Gen ChunkedEvents) -> Gen ChunkedEvents
forall a b. (a -> b) -> a -> b
$ \Int
n -> do
    -- ensure at least one event
    let total :: EventId
total = SimpleId -> EventId
forall a. Num a => SimpleId -> a
fromInteger (SimpleId -> EventId) -> (Int -> SimpleId) -> Int -> EventId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int -> SimpleId
forall a. Integral a => a -> SimpleId
toInteger (Int -> EventId) -> Int -> EventId
forall a b. (a -> b) -> a -> b
$ Int
1 Int -> Int -> Int
forall a. Ord a => a -> a -> a
`max` Int
n
    let events :: [TrivialEvent]
events = (EventId -> TrivialEvent) -> [EventId] -> [TrivialEvent]
forall a b. (a -> b) -> [a] -> [b]
map EventId -> TrivialEvent
TrivialEvent [EventId
1 .. EventId
total]
    [[TrivialEvent]]
chunks <- ([TrivialEvent], [[TrivialEvent]]) -> Gen [[TrivialEvent]]
forall a. ([a], [[a]]) -> Gen [[a]]
chunkRandomly ([TrivialEvent]
events, [])
    ChunkedEvents -> Gen ChunkedEvents
forall a. a -> Gen a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([[TrivialEvent]] -> ChunkedEvents
ChunkedEvents [[TrivialEvent]]
chunks)
   where
    chunkRandomly :: ([a], [[a]]) -> Gen [[a]]
    chunkRandomly :: forall a. ([a], [[a]]) -> Gen [[a]]
chunkRandomly = \case
      ([], [[a]]
acc) -> [[a]] -> Gen [[a]]
forall a. a -> Gen a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([[a]] -> [[a]]
forall a. [a] -> [a]
reverse [[a]]
acc)
      ([a]
xs, [[a]]
acc) -> do
        -- allow random-sized chunks, including empty ones
        Int
chunkSize <- (Int, Int) -> Gen Int
forall a. Random a => (a, a) -> Gen a
choose (Int
0, [a] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [a]
xs)
        let ([a]
chunk, [a]
rest) = Int -> [a] -> ([a], [a])
forall a. Int -> [a] -> ([a], [a])
splitAt Int
chunkSize [a]
xs
        ([a], [[a]]) -> Gen [[a]]
forall a. ([a], [[a]]) -> Gen [[a]]
chunkRandomly ([a]
rest, [a]
chunk [a] -> [[a]] -> [[a]]
forall a. a -> [a] -> [a]
: [[a]]
acc)

instance HasEventId TrivialEvent where
  getEventId :: TrivialEvent -> EventId
getEventId (TrivialEvent EventId
w) = EventId
w

trivialCheckpoint :: [TrivialEvent] -> TrivialEvent
trivialCheckpoint :: [TrivialEvent] -> TrivialEvent
trivialCheckpoint = [TrivialEvent] -> TrivialEvent
forall a (f :: * -> *). (Foldable f, Num a) => f a -> a
sum

mkAggregator :: IsChainState tx => NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator :: forall tx.
IsChainState tx =>
NodeState tx -> StateEvent tx -> NodeState tx
mkAggregator NodeState tx
s StateEvent{StateChanged tx
$sel:stateChanged:StateEvent :: forall tx. StateEvent tx -> StateChanged tx
stateChanged :: StateChanged tx
stateChanged} = NodeState tx -> StateChanged tx -> NodeState tx
forall tx.
IsChainState tx =>
NodeState tx -> StateChanged tx -> NodeState tx
aggregateNodeState NodeState tx
s StateChanged tx
stateChanged