Enhance NodeState to keep track of next EventId

This way we keep it in memory, but do not make the HeadLogic depend on
or require the current / next EventId.

Author: ch1bo

hydra-node/src/Hydra/Node.hs

@@ -15,6 +15,7 @@ import Control.Concurrent.Class.MonadSTM (
   labelTVarIO,
   newTVarIO,
   stateTVar,
+  writeTVar,
  )
 import Control.Monad.Trans.Writer (execWriter, tell)
 import Hydra.API.ClientInput (ClientInput)
@@ -33,7 +34,7 @@ import Hydra.Chain.Direct.Tx (verificationKeyToOnChainId)
 import Hydra.Chain.Direct.Util (readFileTextEnvelopeThrow)
 import Hydra.Crypto (AsType (AsHydraKey))
 import Hydra.Environment (Environment (..))
-import Hydra.Events (EventSink (..), EventSource (..), StateEvent (..), putEventToSinks, putEventsToSinks, stateChanged)
+import Hydra.Events (EventId, EventSink (..), EventSource (..), StateEvent (..), getEventId, putEventToSinks, putEventsToSinks, stateChanged)
 import Hydra.HeadLogic (
   Effect (..),
   HeadState (..),
@@ -179,6 +180,7 @@ hydrate ::
   m (WetHydraNode tx m)
 hydrate eventSource eventSinks dryNode = do
   events <- getEvents eventSource
+  let lastSeenEventId = getEventId . last <$> nonEmpty events
   traceWith tracer LoadedState{numberOfEvents = fromIntegral $ length events}
   let headState = recoverState initialState (stateChanged <$> events)
       chainStateHistory = recoverChainStateHistory initialChainState (stateChanged <$> events)
@@ -187,7 +189,7 @@ hydrate eventSource eventSinks dryNode = do
   -- deliver to sinks per spec, deduplication is handled by the sinks
   -- FIXME(Elaine): persistence currently not handling duplication, so this relies on not providing the eventSource's sink as an arg here
   putEventsToSinks eventSinks events
-  nodeState <- createNodeState headState
+  nodeState <- createNodeState lastSeenEventId headState
   inputQueue <- createInputQueue
   pure
     WetHydraNode
@@ -312,13 +314,17 @@ processNextInput HydraNode{nodeState, ledger, env} e =
 
   computeOutcome = HeadLogic.update env ledger
 
-processStateChanges :: Monad m => HydraNode tx m -> [StateChanged tx] -> m ()
+processStateChanges :: MonadSTM m => HydraNode tx m -> [StateChanged tx] -> m ()
 processStateChanges node stateChanges = do
-  -- TODO: obviously don't do this
-  let events = zipWith (\eventId stateChanged -> StateEvent{eventId, stateChanged}) [0 ..] stateChanges
+  events <- atomically . forM stateChanges $ \stateChanged -> do
+    eventId <- getNextEventId
+    pure StateEvent{eventId, stateChanged}
   forM_ events $ putEventToSinks eventSinks
  where
-  HydraNode{eventSinks} = node
+  HydraNode
+    { eventSinks
+    , nodeState = NodeState{getNextEventId}
+    } = node
 
 processEffects ::
   ( MonadAsync m
@@ -358,17 +364,29 @@ processEffects node tracer inputId effects = do
 data NodeState tx m = NodeState
   { modifyHeadState :: forall a. (HeadState tx -> (a, HeadState tx)) -> STM m a
   , queryHeadState :: STM m (HeadState tx)
+  , getNextEventId :: STM m EventId
   }
 
 -- | Initialize a new 'NodeState'.
-createNodeState :: MonadLabelledSTM m => HeadState tx -> m (NodeState tx m)
-createNodeState initialState = do
-  tv <- newTVarIO initialState
-  labelTVarIO tv "node-state"
+createNodeState ::
+  MonadLabelledSTM m =>
+  -- | Last seen 'EventId'.
+  Maybe EventId ->
+  HeadState tx ->
+  m (NodeState tx m)
+createNodeState lastSeenEventId initialState = do
+  nextEventIdV <- newTVarIO $ maybe 0 (+ 1) lastSeenEventId
+  labelTVarIO nextEventIdV "next-event-id"
+  hs <- newTVarIO initialState
+  labelTVarIO hs "head-state"
   pure
     NodeState
-      { modifyHeadState = stateTVar tv
-      , queryHeadState = readTVar tv
+      { modifyHeadState = stateTVar hs
+      , queryHeadState = readTVar hs
+      , getNextEventId = do
+          eventId <- readTVar nextEventIdV
+          writeTVar nextEventIdV $ eventId + 1
+          pure eventId
       }
 
 -- * Logging

hydra-node/test/Hydra/BehaviorSpec.hs

@@ -746,7 +746,7 @@ withHydraNode ::
 withHydraNode signingKey otherParties chain action = do
   outputs <- atomically newTQueue
   outputHistory <- newTVarIO mempty
-  nodeState <- createNodeState $ Idle IdleState{chainState = SimpleChainState{slot = ChainSlot 0}}
+  nodeState <- createNodeState Nothing $ Idle IdleState{chainState = SimpleChainState{slot = ChainSlot 0}}
   node <- createHydraNode traceInIOSim simpleLedger nodeState signingKey otherParties outputs outputHistory chain testContestationPeriod
   withAsync (runHydraNode node) $ \_ ->
     action (createTestHydraClient outputs outputHistory node nodeState)

hydra-node/test/Hydra/Model.hs

@@ -587,7 +587,7 @@ seedWorld seedKeys seedCP futureCommits = do
       labelTQueueIO outputs ("outputs-" <> shortLabel hsk)
       outputHistory <- newTVarIO []
       labelTVarIO outputHistory ("history-" <> shortLabel hsk)
-      nodeState <- createNodeState $ HeadState.Idle IdleState{chainState = initialChainState}
+      nodeState <- createNodeState Nothing $ HeadState.Idle IdleState{chainState = initialChainState}
       node <- createHydraNode (contramap Node tr) ledger nodeState hsk otherParties outputs outputHistory mockChain seedCP
       let testClient = createTestHydraClient outputs outputHistory node nodeState
       nodeThread <- async $ labelThisThread ("node-" <> shortLabel hsk) >> runHydraNode node