Crudely disable API server history

Author: ch1bo

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

@@ -7,7 +7,7 @@ import Hydra.Prelude hiding (TVar, readTVar, seq)
 import Cardano.Ledger.Core (PParams)
 import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Control.Concurrent.STM.TChan (newBroadcastTChanIO, writeTChan)
-import Control.Concurrent.STM.TVar (modifyTVar', newTVarIO)
+import Control.Concurrent.STM.TVar (modifyTVar', newTVar, newTVarIO, readTVar)
 import Control.Exception (IOException)
 import Hydra.API.APIServerLog (APIServerLog (..))
 import Hydra.API.ClientInput (ClientInput)
@@ -79,17 +79,19 @@ withAPIServer ::
 withAPIServer config party persistence tracer chain pparams callback action =
   handle onIOException $ do
     responseChannel <- newBroadcastTChanIO
+    -- Intialize our read models from stored events
+    -- NOTE: we do not keep the stored events around in memory
     timedOutputEvents <- loadAll
-
-    -- Intialize our read model from stored events
     headStatusP <- mkProjection Idle (output <$> timedOutputEvents) projectHeadStatus
     snapshotUtxoP <- mkProjection Nothing (output <$> timedOutputEvents) projectSnapshotUtxo
     headIdP <- mkProjection Nothing (output <$> timedOutputEvents) projectInitializingHeadId
 
-    -- NOTE: we need to reverse the list because we store history in a reversed
-    -- list in memory but in order on disk
-    -- FIXME: always growing
-    history <- newTVarIO (reverse timedOutputEvents)
+    nextSeqVar <- newTVarIO 0
+    let nextSeq = atomically $ do
+      seq <- readTVar nextSeqVar
+      modifyTVar' nextSeqVar (+ 1)
+      pure seq
+
     (notifyServerRunning, waitForServerRunning) <- setupServerNotification
 
     let serverSettings =
@@ -106,15 +108,15 @@ withAPIServer config party persistence tracer chain pparams callback action =
             . simpleCors
             $ websocketsOr
               defaultConnectionOptions
-              (wsApp party tracer history callback headStatusP snapshotUtxoP responseChannel)
+              (wsApp party tracer nextSeq callback headStatusP snapshotUtxoP responseChannel)
               (httpApp tracer chain pparams (atomically $ getLatest headIdP) (atomically $ getLatest snapshotUtxoP) callback)
       )
       ( do
           waitForServerRunning
           action $
             Server
               { sendOutput = \output -> do
-                  timedOutput <- appendToHistory history output
+                  timedOutput <- persistOutput nextSeq output
                   atomically $ do
                     update headStatusP output
                     update snapshotUtxoP output
@@ -125,7 +127,7 @@ withAPIServer config party persistence tracer chain pparams callback action =
  where
   APIServerConfig{host, port, tlsCertPath, tlsKeyPath} = config
 
-  PersistenceIncremental{loadAll, append} = persistence
+  PersistenceIncremental{loadAll} = persistence
 
   startServer settings app =
     case (tlsCertPath, tlsKeyPath) of
@@ -139,13 +141,11 @@ withAPIServer config party persistence tracer chain pparams callback action =
       _ ->
         runSettings settings app
 
-  appendToHistory history output = do
+  persistOutput nextSeq output = do
     time <- getCurrentTime
     timedOutput <- atomically $ do
-      seq <- nextSequenceNumber history
-      let timedOutput = TimedServerOutput{output, time, seq}
-      modifyTVar' history (timedOutput :)
-      pure timedOutput
+      seq <- nextSeq
+      pure TimedServerOutput{output, time, seq}
     append timedOutput
     pure timedOutput
 

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

@@ -49,7 +49,8 @@ wsApp ::
   IsChainState tx =>
   Party ->
   Tracer IO APIServerLog ->
-  TVar [TimedServerOutput tx] ->
+  -- | Get next sequence number.
+  STM IO Natural ->
   (ClientInput tx -> IO ()) ->
   -- | Read model to enhance 'Greetings' messages with 'HeadStatus'.
   Projection STM.STM (ServerOutput tx) HeadStatus ->
@@ -58,18 +59,19 @@ wsApp ::
   TChan (TimedServerOutput tx) ->
   PendingConnection ->
   IO ()
-wsApp party tracer history callback headStatusP snapshotUtxoP responseChannel pending = do
+wsApp party tracer nextSeq callback headStatusP snapshotUtxoP responseChannel pending = do
   traceWith tracer NewAPIConnection
   let path = requestPath $ pendingRequest pending
   queryParams <- uriQuery <$> mkURIBs path
   con <- acceptRequest pending
   chan <- STM.atomically $ dupTChan responseChannel
 
+  -- FIXME: No support of history forwarding anymore (disabled because of memory growing too much)
   -- api client can decide if they want to see the past history of server outputs
-  unless (shouldNotServeHistory queryParams) $
-    forwardHistory con
+  -- unless (shouldNotServeHistory queryParams) $
+  --   forwardHistory con
 
-  forwardGreetingOnly con
+  sendGreetings con
 
   let outConfig = mkServerOutputConfig queryParams
 
@@ -79,8 +81,8 @@ wsApp party tracer history callback headStatusP snapshotUtxoP responseChannel pe
   -- NOTE: We will add a 'Greetings' message on each API server start. This is
   -- important to make sure the latest configured 'party' is reaching the
   -- client.
-  forwardGreetingOnly con = do
-    seq <- atomically $ nextSequenceNumber history
+  sendGreetings con = do
+    seq <- atomically nextSeq
     headStatus <- atomically getLatestHeadStatus
     snapshotUtxo <- atomically getLatestSnapshotUtxo
     time <- getCurrentTime
@@ -114,11 +116,11 @@ wsApp party tracer history callback headStatusP snapshotUtxoP responseChannel pe
         queryP = QueryParam k v
      in if queryP `elem` qp then WithoutUTxO else WithUTxO
 
-  shouldNotServeHistory qp =
-    flip any qp $ \case
-      (QueryParam key val)
-        | key == [queryKey|history|] -> val == [queryValue|no|]
-      _other -> False
+  -- shouldNotServeHistory qp =
+  --   flip any qp $ \case
+  --     (QueryParam key val)
+  --       | key == [queryKey|history|] -> val == [queryValue|no|]
+  --     _other -> False
 
   sendOutputs chan con outConfig = forever $ do
     response <- STM.atomically $ readTChan chan
@@ -139,18 +141,12 @@ wsApp party tracer history callback headStatusP snapshotUtxoP responseChannel pe
         -- message to memory
         let clientInput = decodeUtf8With lenientDecode $ toStrict msg
         time <- getCurrentTime
-        seq <- atomically $ nextSequenceNumber history
+        seq <- atomically nextSeq
         let timedOutput = TimedServerOutput{output = InvalidInput @tx e clientInput, time, seq}
         sendTextData con $ Aeson.encode timedOutput
         traceWith tracer (APIInvalidInput e clientInput)
 
-  forwardHistory con = do
-    hist <- STM.atomically (readTVar history)
-    let encodeAndReverse xs serverOutput = Aeson.encode serverOutput : xs
-    sendTextDatas con $ foldl' encodeAndReverse [] hist
-
-nextSequenceNumber :: TVar [TimedServerOutput tx] -> STM.STM Natural
-nextSequenceNumber historyList =
-  STM.readTVar historyList >>= \case
-    [] -> pure 0
-    (TimedServerOutput{seq} : _) -> pure (seq + 1)
+  -- forwardHistory con = do
+  --   hist <- STM.atomically (readTVar history)
+  --   let encodeAndReverse xs serverOutput = Aeson.encode serverOutput : xs
+  --   sendTextDatas con $ foldl' encodeAndReverse [] hist