Event log rotation (#1997)

<!-- Describe your change here -->

- Event log rotation enable hydra heads to resume faster by reducing
replay time.

- Rotating an event log means replacing the current state file with a
new one that starts from a checkpoint event, while moving the old file
to the next state-logId.

- A Checkpoint event captures the latest HeadState, computed by
aggregating all previous StateChanged events.

- File-based persistence rotates event log files using an increasing
index named logId, based on latest StateChanged event id + 1.

- Added a new run option to enable rotation after a given number of
events:
_"The number of Hydra events to trigger rotation (default: no
rotation)"_

- On startup, depending on the rotation config used, the event log file
might be rotated and the `logId` index will be incremented.

- Added new server output to allow 3rd party agents to detect the
checkpoint and trigger any appropriate archival / backup / cleanup
needed, without interrupting the hydra head.

---

<!-- Consider each and tick it off one way or the other -->
* [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

---------

Co-authored-by: Sebastian Nagel <[email protected]>
Co-authored-by: Sebastian Nagel <[email protected]>

Author: ffakenz

CHANGELOG.md

@@ -17,6 +17,8 @@ changes.
 - Remove runtime dependency to `etcd` by embedding and shipping it with `hydra-node`.
   - New option `--use-system-etcd` to prefer the system etcd instead of the embedded one.
 
+- Add file-based event log rotation support via optional `--persistence-rotate-after` command line option.
+
 - **BREAKING** Update scripts to plutus 1.45.0.0.
 
 - Hydra will now store etcd cluster information on the filesystem in directories content-addressed

docs/docs/dev/architecture/index.md

@@ -53,7 +53,7 @@ The `hydra-node` component exposes an [asynchronous API](https://hydra.family/he
 
 ### Persistence
 
-All API server outputs and the `hydra-node` state are preserved on disk. The persistence layer is responsible for loading historical messages and the Hydra state from disk, as well as storing them. Currently, there hasn't been a need to increase the complexity of this layer or use a database.
+The `hydra-node` state is preserved on disk. The persistence layer is responsible for loading historical messages and Hydra state from disk, as well as storing them in so-called event log files. Depending on the rotation configuration used at startup, these event log files will be rotated to improve restart times. So far, there hasn’t been a need to increase the complexity of this layer or to use a database.
 
 ### Logging
 

docs/docs/known-issues.md

@@ -47,9 +47,9 @@ here: [Etcd Configuration](configuration#networking-configuring-the-limits-of-et
 
 If the hydra-node has breaking changes in regards to reading the files it stores in the `persistence` folder, it used to be recommended to just delete the entire folder.
 
-Now, because of etcd, it is important to only delete the `hydra-node` specific files; not the files associated with `etcd`. In particular you may like to delete the following file:
+Now, because of etcd, it is important to only delete the `hydra-node` specific files; not the files associated with `etcd`. In particular you may like to delete the following files:
 
-- `persistence/state`
+- `persistence/state*`
 
 Note that, as with any adjustments of this kind, it is good practice to make a backup first!
 

hydra-cluster/src/Hydra/Cluster/Scenarios.hs

@@ -152,7 +152,7 @@ import Network.HTTP.Req (
 import Network.HTTP.Simple (getResponseBody, httpJSON, setRequestBodyJSON)
 import Network.HTTP.Types (urlEncode)
 import System.FilePath ((</>))
-import System.Process (proc, readCreateProcessWithExitCode)
+import System.Process (callProcess)
 import Test.Hydra.Tx.Fixture (testNetworkId)
 import Test.Hydra.Tx.Gen (genKeyPair)
 import Test.QuickCheck (choose, elements, generate)
@@ -401,9 +401,9 @@ nodeReObservesOnChainTxs tracer workDir cardanoNode hydraScriptsTxId = do
         <&> modifyConfig (\config -> config{startChainFrom = Just tip})
 
     withTempDir "blank-state" $ \tmpDir -> do
-      void $ readCreateProcessWithExitCode (proc "cp" ["-r", workDir </> "state-2", tmpDir]) ""
-      void $ readCreateProcessWithExitCode (proc "rm" ["-rf", tmpDir </> "state-2" </> "state"]) ""
-      void $ readCreateProcessWithExitCode (proc "rm" ["-rf", tmpDir </> "state-2" </> "last-known-revision"]) ""
+      callProcess "cp" ["-r", workDir </> "state-2", tmpDir]
+      callProcess "rm" ["-rf", tmpDir </> "state-2" </> "state*"]
+      callProcess "rm" ["-rf", tmpDir </> "state-2" </> "last-known-revision"]
       withHydraNode hydraTracer bobChainConfigFromTip tmpDir 2 bobSk [aliceVk] [1] $ \n2 -> do
         -- Also expect to see past server outputs replayed
         headId2 <- waitMatch 5 n2 $ headIsInitializingWith (Set.fromList [alice, bob])

hydra-cluster/src/HydraNode.hs

@@ -405,6 +405,7 @@ prepareHydraNode chainConfig workDir hydraNodeId hydraSKey hydraVKeys allNodeIds
       , hydraSigningKey
       , hydraVerificationKeys
       , persistenceDir = stateDir
+      , persistenceRotateAfter = Nothing
       , chainConfig
       , whichEtcd = EmbeddedEtcd
       , ledgerConfig =
@@ -531,6 +532,7 @@ withHydraNode tracer chainConfig workDir hydraNodeId hydraSKey hydraVKeys allNod
                 , hydraSigningKey
                 , hydraVerificationKeys
                 , persistenceDir = stateDir
+                , persistenceRotateAfter = Nothing
                 , chainConfig
                 , whichEtcd = EmbeddedEtcd
                 , ledgerConfig =

hydra-cluster/test/Test/EndToEndSpec.hs

@@ -20,9 +20,11 @@ import CardanoNode (
   withCardanoNodeDevnet,
  )
 import Control.Lens ((^..), (^?))
+import Control.Monad (foldM_)
 import Data.Aeson (Result (..), Value (Null, Object, String), fromJSON, object, (.=))
 import Data.Aeson qualified as Aeson
 import Data.Aeson.Lens (AsJSON (_JSON), key, values, _JSON)
+import Data.Aeson.Types (parseMaybe)
 import Data.ByteString qualified as BS
 import Data.List qualified as List
 import Data.Map qualified as Map
@@ -87,6 +89,7 @@ import HydraNode (
   getSnapshotUTxO,
   input,
   output,
+  prepareHydraNode,
   requestCommitTx,
   send,
   waitFor,
@@ -95,6 +98,7 @@ import HydraNode (
   waitMatch,
   withHydraCluster,
   withHydraNode,
+  withPreparedHydraNode,
  )
 import System.Directory (removeDirectoryRecursive, removeFile)
 import System.FilePath ((</>))
@@ -158,6 +162,57 @@ spec = around (showLogsOnFailure "EndToEndSpec") $ do
           waitMatch 10 node $ \v -> do
             guard $ v ^? key "tag" == Just "SnapshotConfirmed"
 
+    it "rotates persistence on start up" $ \tracer -> do
+      withClusterTempDir $ \tmpDir -> do
+        (aliceCardanoVk, aliceCardanoSk) <- keysFor Alice
+        initialUTxO <- generate $ genUTxOFor aliceCardanoVk
+        Aeson.encodeFile (tmpDir </> "utxo.json") initialUTxO
+        let offlineConfig =
+              Offline
+                OfflineChainConfig
+                  { offlineHeadSeed = "test"
+                  , initialUTxOFile = tmpDir </> "utxo.json"
+                  , ledgerGenesisFile = Nothing
+                  }
+        -- Start a hydra-node in offline mode and submit several self-txs
+        withHydraNode (contramap FromHydraNode tracer) offlineConfig tmpDir 1 aliceSk [] [] $ \node -> do
+          foldM_
+            ( \utxo i -> do
+                let Just (aliceTxIn, aliceTxOut) = UTxO.find (isVkTxOut aliceCardanoVk) utxo
+                let Right selfTx =
+                      mkSimpleTx
+                        (aliceTxIn, aliceTxOut)
+                        (mkVkAddress testNetworkId aliceCardanoVk, txOutValue aliceTxOut)
+                        aliceCardanoSk
+                send node $ input "NewTx" ["transaction" .= selfTx]
+                waitMatch 10 node $ \v -> do
+                  guard $ v ^? key "tag" == Just "SnapshotConfirmed"
+                  guard $ v ^? key "snapshot" . key "number" == Just (toJSON (i :: Integer))
+                  v ^? key "snapshot" . key "utxo" >>= parseMaybe parseJSON
+            )
+            initialUTxO
+            [1 .. (200 :: Integer)]
+
+        -- Measure restart time
+        t0 <- getCurrentTime
+        diff1 <- withHydraNode (contramap FromHydraNode tracer) offlineConfig tmpDir 1 aliceSk [] [] $ \_ -> do
+          t1 <- getCurrentTime
+          let diff = diffUTCTime t1 t0
+          pure diff
+
+        -- Measure restart after rotation
+        options <- prepareHydraNode offlineConfig tmpDir 1 aliceSk [] [] id
+        let options' = options{persistenceRotateAfter = Just 10}
+        t1 <- getCurrentTime
+        diff2 <- withPreparedHydraNode (contramap FromHydraNode tracer) tmpDir 1 options' $ \_ -> do
+          t2 <- getCurrentTime
+          let diff = diffUTCTime t2 t1
+          pure diff
+
+        unless (diff2 < diff1 * 0.9) $
+          failure $
+            "Expected to start up 10% quicker than original " <> show diff1 <> ", but it took " <> show diff2
+
     it "supports multi-party networked heads" $ \tracer -> do
       withClusterTempDir $ \tmpDir -> do
         (aliceCardanoVk, aliceCardanoSk) <- keysFor Alice