Streaming events (#1808)

Changes I made originally for the `hydra-doom` project to load all
events in `state` (12GB+ in that use case) with constant memory using
`conduit` streams.

~~There is a big TODO on this: `IOSim s` does not have a `MonadUnliftIO`
instance and its not impossible to have one. We need to change the
interface further such that we can compose `createHydraNode` and
`hydrate` functions with in-memory `EventSource` variants.~~

~~A bit of a wart: The `mkProjection` does run the conduit for each
projection, instead we should run the conduit once and build the
projected in-memory read/query model once.~~

Also, the change here is not constant-memory as the `ServerOutput`
history is still kept fully in memory. But this is a different story and
should be covered by #1618

---

* [x] CHANGELOG updated or not needed
* [x] Documentation updated or not needed
* [x] Haddocks updated or not needed
* [x] No new TODOs introduced or explained herafter

Author: v0d1ch

CHANGELOG.md

@@ -10,7 +10,10 @@ changes.
 
 ## [0.20.1] - UNRELEASED
 
+- Stream historical data from disk in the hydra-node API server.
+
 - Record used and free memory when running `bench-e2e` benchmark.
+
 - Submit observations to a `hydra-explorer` via optional `--explorer` option.
 
 - Tested with `cardano-node 10.2` and `cardano-cli 10.3.0.0`.

hydra-node/hydra-node.cabal

@@ -118,6 +118,7 @@ library
     , cardano-slotting
     , cardano-strict-containers
     , cborg
+    , conduit
     , containers
     , contra-tracer
     , cryptonite
@@ -340,6 +341,7 @@ test-suite tests
     , cardano-slotting
     , cardano-strict-containers
     , cborg
+    , conduit
     , containers
     , contra-tracer
     , directory

hydra-node/json-schemas/logs.yaml

@@ -894,21 +894,45 @@ definitions:
             minimum: 0
           input:
             "$ref": "logs.yaml#/definitions/Input"
+      - title: LoadingState
+        description: >-
+          Loading state events from persistence.
+        type: object
+        required:
+          - tag
+        properties:
+          tag:
+            type: string
+            enum: ["LoadingState"]
+      - title: ReplayingState
+        description: >-
+          Replaying state events from persistence.
+        type: object
+        required:
+          - tag
+        properties:
+          tag:
+            type: string
+            enum: ["ReplayingState"]
       - title: LoadedState
         description: >-
           Loaded state events from persistence.
         type: object
         additionalProperties: false
         required:
           - tag
-          - numberOfEvents
+          - lastEventId
+          - headState
         properties:
           tag:
             type: string
             enum: ["LoadedState"]
-          numberOfEvents:
-            type: integer
-            minimum: 0
+          lastEventId:
+            oneOf:
+              - type: "null"
+              - type: integer
+          headState:
+            $ref: "logs.yaml#/definitions/HeadState"
       - title: Misconfiguration
         description: >-
           Hydra node detected a difference between loaded state and the node arguments.

hydra-node/src/Hydra/API/Projection.hs

@@ -17,7 +17,7 @@ module Hydra.API.Projection where
 
 import Hydra.Prelude
 
-import Control.Concurrent.Class.MonadSTM (modifyTVar', newTVar)
+import Control.Concurrent.Class.MonadSTM (modifyTVar', newTVarIO)
 
 -- | 'Projection' type used to alter/project the API output to suit the client needs.
 data Projection stm event model = Projection
@@ -33,19 +33,16 @@ data Projection stm event model = Projection
 mkProjection ::
   MonadSTM m =>
   model ->
-  [event] ->
   -- | Projection function
   (model -> event -> model) ->
   m (Projection (STM m) event model)
-mkProjection startingModel events project =
-  atomically $ do
-    tv <- newTVar startingModel
-    mapM_ (update tv) events
-    pure
-      Projection
-        { getLatest = readTVar tv
-        , update = update tv
-        }
+mkProjection startingModel project = do
+  tv <- newTVarIO startingModel
+  pure
+    Projection
+      { getLatest = readTVar tv
+      , update = update tv
+      }
  where
   update tv event =
     modifyTVar' tv $ \m ->

hydra-node/src/Hydra/API/Server.hs

@@ -2,13 +2,15 @@
 
 module Hydra.API.Server where
 
-import Hydra.Prelude hiding (TVar, readTVar, seq)
+import Hydra.Prelude hiding (TVar, mapM_, readTVar, seq)
 
 import Cardano.Ledger.Core (PParams)
+import Conduit (runConduitRes, sinkList, (.|))
 import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Control.Concurrent.STM.TChan (newBroadcastTChanIO, writeTChan)
 import Control.Concurrent.STM.TVar (modifyTVar', newTVarIO)
 import Control.Exception (IOException)
+import Data.Conduit.Combinators (iterM)
 import Hydra.API.APIServerLog (APIServerLog (..))
 import Hydra.API.ClientInput (ClientInput)
 import Hydra.API.HTTPServer (httpApp)
@@ -89,18 +91,28 @@ withAPIServer ::
 withAPIServer config env party persistence tracer chain pparams serverOutputFilter callback action =
   handle onIOException $ do
     responseChannel <- newBroadcastTChanIO
-    timedOutputEvents <- loadAll
-
-    -- Intialize our read model from stored events
-    headStatusP <- mkProjection Idle (output <$> timedOutputEvents) projectHeadStatus
-    snapshotUtxoP <- mkProjection Nothing (output <$> timedOutputEvents) projectSnapshotUtxo
-    commitInfoP <- mkProjection CannotCommit (output <$> timedOutputEvents) projectCommitInfo
-    headIdP <- mkProjection Nothing (output <$> timedOutputEvents) projectInitializingHeadId
-    pendingDepositsP <- mkProjection [] (output <$> timedOutputEvents) projectPendingDeposits
-
+    -- Intialize our read models from stored events
+    -- NOTE: we do not keep the stored events around in memory
+    headStatusP <- mkProjection Idle projectHeadStatus
+    snapshotUtxoP <- mkProjection Nothing projectSnapshotUtxo
+    commitInfoP <- mkProjection CannotCommit projectCommitInfo
+    headIdP <- mkProjection Nothing projectInitializingHeadId
+    pendingDepositsP <- mkProjection [] projectPendingDeposits
+    loadedHistory <-
+      runConduitRes $
+        source
+          -- .| mapC output
+          .| iterM (lift . atomically . update headStatusP . output)
+          .| iterM (lift . atomically . update snapshotUtxoP . output)
+          .| iterM (lift . atomically . update commitInfoP . output)
+          .| iterM (lift . atomically . update headIdP . output)
+          .| iterM (lift . atomically . update pendingDepositsP . output)
+          -- FIXME: don't load whole history into memory
+          .| sinkList
+
+    history <- newTVarIO loadedHistory
     -- NOTE: we need to reverse the list because we store history in a reversed
     -- list in memory but in order on disk
-    history <- newTVarIO (reverse timedOutputEvents)
     (notifyServerRunning, waitForServerRunning) <- setupServerNotification
 
     let serverSettings =
@@ -138,7 +150,7 @@ withAPIServer config env party persistence tracer chain pparams serverOutputFilt
  where
   APIServerConfig{host, port, tlsCertPath, tlsKeyPath} = config
 
-  PersistenceIncremental{loadAll, append} = persistence
+  PersistenceIncremental{source, append} = persistence
 
   startServer settings app =
     case (tlsCertPath, tlsKeyPath) of

hydra-node/src/Hydra/Events.hs

@@ -15,6 +15,7 @@ module Hydra.Events where
 
 import Hydra.Prelude
 
+import Conduit (ConduitT, MonadUnliftIO, ResourceT, runResourceT, sourceToList)
 import Hydra.Chain.ChainState (IsChainState)
 import Hydra.HeadLogic.Outcome (StateChanged)
 import Hydra.Tx.IsTx (ArbitraryIsTx)
@@ -25,10 +26,14 @@ class HasEventId a where
   getEventId :: a -> EventId
 
 newtype EventSource e m = EventSource
-  { getEvents :: HasEventId e => m [e]
-  -- ^ Retrieve all events from the event source.
+  { sourceEvents :: HasEventId e => ConduitT () e (ResourceT m) ()
+  -- ^ Stream all events from the event source.
   }
 
+-- | Retrieve all events from the event source as a list.
+getEvents :: (HasEventId e, MonadUnliftIO m) => EventSource e m -> m [e]
+getEvents EventSource{sourceEvents} = runResourceT $ sourceToList sourceEvents
+
 newtype EventSink e m = EventSink
   { putEvent :: HasEventId e => e -> m ()
   -- ^ Send a single event to the event sink.

hydra-node/src/Hydra/Events/FileBased.hs

@@ -5,6 +5,7 @@ module Hydra.Events.FileBased where
 
 import Hydra.Prelude
 
+import Conduit (mapMC, (.|))
 import Control.Concurrent.Class.MonadSTM (newTVarIO, writeTVar)
 import Hydra.Chain.ChainState (IsChainState)
 import Hydra.Events (EventSink (..), EventSource (..), StateEvent (..))
@@ -29,7 +30,7 @@ eventPairFromPersistenceIncremental ::
   (IsChainState tx, MonadSTM m) =>
   PersistenceIncremental (PersistedStateChange tx) m ->
   m (EventSource (StateEvent tx) m, EventSink (StateEvent tx) m)
-eventPairFromPersistenceIncremental PersistenceIncremental{append, loadAll} = do
+eventPairFromPersistenceIncremental PersistenceIncremental{append, source} = do
   eventIdV <- newTVarIO Nothing
   let
     getLastSeenEventId = readTVar eventIdV
@@ -41,17 +42,18 @@ eventPairFromPersistenceIncremental PersistenceIncremental{append, loadAll} = do
       maybe 0 (+ 1) <$> readTVar eventIdV
 
     -- Keep track of the last seen event id when loading
-    getEvents = do
-      items <- loadAll
-      atomically . forM items $ \i -> do
-        event <- case i of
-          New e -> pure e
-          Legacy sc -> do
-            eventId <- getNextEventId
-            pure $ StateEvent eventId sc
-
-        setLastSeenEventId event
-        pure event
+    sourceEvents =
+      source
+        .| mapMC
+          ( \i -> lift . atomically $ do
+              event <- case i of
+                New e -> pure e
+                Legacy sc -> do
+                  eventId <- getNextEventId
+                  pure $ StateEvent eventId sc
+              setLastSeenEventId event
+              pure event
+          )
 
     -- Filter events that are already stored
     putEvent e@StateEvent{eventId} = do
@@ -65,7 +67,7 @@ eventPairFromPersistenceIncremental PersistenceIncremental{append, loadAll} = do
       append (New e)
       atomically $ setLastSeenEventId e
 
-  pure (EventSource{getEvents}, EventSink{putEvent})
+  pure (EventSource{sourceEvents}, EventSink{putEvent})
 
 -- | Internal data type used by 'createJSONFileEventSourceAndSink' to be
 -- compatible with plain usage of 'PersistenceIncrementa' using plain

hydra-node/src/Hydra/HeadLogic.hs

@@ -34,7 +34,6 @@ import Hydra.Chain (
   ChainStateHistory,
   OnChainTx (..),
   PostChainTx (..),
-  initHistory,
   pushNewState,
   rollbackHistory,
  )
@@ -1666,48 +1665,34 @@ aggregateState ::
   Outcome tx ->
   HeadState tx
 aggregateState s outcome =
-  recoverState s $ collectStateChanged outcome
+  foldl' aggregate s $ collectStateChanged outcome
  where
   collectStateChanged = \case
     Error{} -> []
     Wait{stateChanges} -> stateChanges
     Continue{stateChanges} -> stateChanges
 
-recoverChainStateHistory ::
-  (Foldable t, IsChainState tx) =>
-  ChainStateType tx ->
-  t (StateChanged tx) ->
-  ChainStateHistory tx
-recoverChainStateHistory initialChainState =
-  foldl' aggregateChainStateHistory (initHistory initialChainState)
- where
-  aggregateChainStateHistory history = \case
-    HeadInitialized{chainState} -> pushNewState chainState history
-    CommittedUTxO{chainState} -> pushNewState chainState history
-    HeadAborted{chainState} -> pushNewState chainState history
-    HeadOpened{chainState} -> pushNewState chainState history
-    TransactionAppliedToLocalUTxO{} -> history
-    CommitRecovered{} -> history
-    CommitRecorded{} -> history
-    DecommitRecorded{} -> history
-    SnapshotRequestDecided{} -> history
-    SnapshotRequested{} -> history
-    TransactionReceived{} -> history
-    PartySignedSnapshot{} -> history
-    SnapshotConfirmed{} -> history
-    CommitFinalized{} -> history
-    DecommitFinalized{} -> history
-    HeadClosed{chainState} -> pushNewState chainState history
-    HeadContested{chainState} -> pushNewState chainState history
-    HeadIsReadyToFanout{} -> history
-    HeadFannedOut{chainState} -> pushNewState chainState history
-    ChainRolledBack{chainState} ->
-      rollbackHistory (chainStateSlot chainState) history
-    TickObserved{} -> history
-
-recoverState ::
-  (Foldable t, IsChainState tx) =>
-  HeadState tx ->
-  t (StateChanged tx) ->
-  HeadState tx
-recoverState = foldl' aggregate
+aggregateChainStateHistory :: IsChainState tx => ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx
+aggregateChainStateHistory history = \case
+  HeadInitialized{chainState} -> pushNewState chainState history
+  CommittedUTxO{chainState} -> pushNewState chainState history
+  HeadAborted{chainState} -> pushNewState chainState history
+  HeadOpened{chainState} -> pushNewState chainState history
+  TransactionAppliedToLocalUTxO{} -> history
+  CommitRecovered{} -> history
+  CommitRecorded{} -> history
+  DecommitRecorded{} -> history
+  SnapshotRequestDecided{} -> history
+  SnapshotRequested{} -> history
+  TransactionReceived{} -> history
+  PartySignedSnapshot{} -> history
+  SnapshotConfirmed{} -> history
+  CommitFinalized{} -> history
+  DecommitFinalized{} -> history
+  HeadClosed{chainState} -> pushNewState chainState history
+  HeadContested{chainState} -> pushNewState chainState history
+  HeadIsReadyToFanout{} -> history
+  HeadFannedOut{chainState} -> pushNewState chainState history
+  ChainRolledBack{chainState} ->
+    rollbackHistory (chainStateSlot chainState) history
+  TickObserved{} -> history

hydra-node/src/Hydra/Network/Reliability.hs

@@ -110,9 +110,10 @@ import Hydra.Logging (traceWith)
 import Hydra.Network (Network (..), NetworkCallback (..), NetworkComponent)
 import Hydra.Network.Authenticate (Authenticated (..))
 import Hydra.Network.Heartbeat (Heartbeat (..), isPing)
-import Hydra.Persistence (Persistence (..), PersistenceIncremental (..))
+import Hydra.Persistence (Persistence (..), PersistenceIncremental (..), loadAll)
 import Hydra.Tx (Party)
 import Test.QuickCheck.Instances.Vector ()
+import UnliftIO (MonadUnliftIO)
 
 data ReliableMsg msg = ReliableMsg
   { knownMessageIds :: Vector Int
@@ -180,7 +181,7 @@ data MessagePersistence m msg = MessagePersistence
 -- NOTE: This handle is returned in the underlying context just for the sake of
 -- convenience.
 mkMessagePersistence ::
-  (MonadThrow m, FromJSON msg, ToJSON msg) =>
+  (MonadUnliftIO m, MonadThrow m, FromJSON msg, ToJSON msg) =>
   Int ->
   PersistenceIncremental (Heartbeat msg) m ->
   Persistence (Vector Int) m ->