Last revision reset (#2137)

noonio · web-flow · commit 1db5ee7064b8 · 2025-07-21T09:43:17.000-06:00
&gt; [!NOTE]
&gt; This is currently on top of the 0.22.2 tag and I could not find a 0.22
release branch to open the PR. We should probably not merge it like
this. Instead I should rebase it to master and we cherry pick it back to
do a 0.22.3 release.

The last-known-revision kept by the Etcd network component may be
incorrect if the etcd cluster compacted this revision while the node was
offline or if the last-known-revision state file was removed.

Both cases can be handled by detecting a failing watch request and at
least using the compactRevision from the response.

This is a somewhat exceptional situation and the node state may be
inconsistent because of this. Hence we also log a warning when this
happens.

---

* [x] CHANGELOG updated
* [ ] Documentation updated or not needed
* [x] Haddocks updated
* [x] No new TODOs introduced or explained herafter
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -8,15 +8,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 As a minor extension, we also keep a semantic version for the `UNRELEASED`
 changes.
 
-## [0.22.2] - 2025.06.30
+## [0.22.3] - UNRELEASED
+
+* Change behavior of `Hydra.Network.Etcd` to fallback to earliest possible
+  revision if `last-known-revision` is missing or too old. This can happen if a
+  node is down for a long time and the `etcd` cluster compacted the last known
+  revision in the meantime
+  [#2136](https://github.com/cardano-scaling/hydra/issues/2136).
+
+## [0.22.2] - 2025-06-30
 
 * Fix wrong hydra-script-tx-ids in networks.json
 
-## [0.22.1] - 2025.06.27
+## [0.22.1] - 2025-06-27
 
 * Fix for bug where node got stalled at `ReplayingState` [#2089](https://github.com/cardano-scaling/hydra/issues/2089)
 
-## [0.22.0] - 2025.06.17
+## [0.22.0] - 2025-06-17
 
 - Tested with `cardano-node 10.1.2` and `cardano-cli 10.1.1.0`.
 
diff --git a/hydra-node/hydra-node.cabal b/hydra-node/hydra-node.cabal
@@ -1,6 +1,6 @@
 cabal-version:      3.0
 name:               hydra-node
-version:            0.22.2
+version:            0.22.3
 synopsis:           The Hydra node
 author:             IOG
 copyright:          2022 IOG
@@ -399,6 +399,7 @@ test-suite tests
     , text
     , time
     , tls
+    , typed-process
     , websockets
 
   build-tool-depends: hspec-discover:hspec-discover
diff --git a/hydra-node/src/Hydra/Network/Etcd.hs b/hydra-node/src/Hydra/Network/Etcd.hs
@@ -22,8 +22,7 @@
 -- only deliver messages that were not seen before. In case we are not connected
 -- to our 'etcd' instance or not enough peers (= on a minority cluster), we
 -- retry sending, but also store messages to broadcast in a 'PersistentQueue',
--- which makes the node resilient against crashes while sending. TODO: Is this
--- needed? performance limitation?
+-- which makes the node resilient against crashes while sending.
 --
 -- Connectivity and compatibility with other nodes on the cluster is tracked
 -- using the key-value service as well:
@@ -92,7 +91,6 @@ import Network.GRPC.Client (
  )
 import Network.GRPC.Client.StreamType.IO (biDiStreaming, nonStreaming)
 import Network.GRPC.Common (GrpcError (..), GrpcException (..), HTTP2Settings (..), NextElem (..), def, defaultHTTP2Settings)
-import Network.GRPC.Common.NextElem (whileNext_)
 import Network.GRPC.Common.Protobuf (Proto (..), Protobuf, defMessage, (.~))
 import Network.GRPC.Etcd (
   Compare'CompareResult (..),
@@ -101,6 +99,7 @@ import Network.GRPC.Etcd (
   Lease,
   Watch,
  )
+import Network.Socket (PortNumber)
 import System.Directory (createDirectoryIfMissing, listDirectory, removeFile)
 import System.Environment.Blank (getEnvironment)
 import System.FilePath (takeDirectory, (</>))
@@ -175,7 +174,7 @@ withEtcdNetwork tracer protocolVersion config callback action = do
         traceWith tracer Reconnecting
         pure $ reconnectPolicy doneVar
 
-  clientHost = Host{hostname = "127.0.0.1", port = clientPort}
+  clientHost = Host{hostname = "127.0.0.1", port = getClientPort config}
 
   grpcServer =
     ServerInsecure $
@@ -185,11 +184,6 @@ withEtcdNetwork tracer protocolVersion config callback action = do
         , addressAuthority = Nothing
         }
 
-  -- NOTE: Offset client port by the same amount as configured 'port' is offset
-  -- from the default '5001'. This will result in the default client port 2379
-  -- be used by default still.
-  clientPort = 2379 + port listen - 5001
-
   traceStderr p NetworkCallback{onConnectivity} =
     forever $ do
       bs <- BS.hGetLine (getStderr p)
@@ -249,6 +243,14 @@ withEtcdNetwork tracer protocolVersion config callback action = do
 
   NetworkConfiguration{persistenceDir, listen, advertise, peers, whichEtcd} = config
 
+-- | Get the client port corresponding to a listen address.
+--
+-- The client port used by the started etcd port is offset by the same amount as
+-- the listen address is offset by the default port 5001. This will result in
+-- the default client port 2379 be used by default still.
+getClientPort :: NetworkConfiguration -> PortNumber
+getClientPort NetworkConfiguration{listen} = 2379 + port listen - 5001
+
 -- | Return the path of the etcd binary. Will either install it first, or just
 -- assume there is one available on the system path.
 getEtcdBinary :: FilePath -> WhichEtcd -> IO FilePath
@@ -297,8 +299,7 @@ checkVersion tracer conn ourVersion NetworkCallback{onConnectivity} = do
         Right theirVersion ->
           unless (theirVersion == ourVersion) $
             onConnectivity VersionMismatch{ourVersion, theirVersion = Just theirVersion}
-    else
-      traceWith tracer $ MatchingProtocolVersion{version = ourVersion}
+    else traceWith tracer $ MatchingProtocolVersion{version = ourVersion}
  where
   versionKey = "version"
 
@@ -371,11 +372,13 @@ waitMessages ::
   NetworkCallback msg IO ->
   IO ()
 waitMessages tracer conn directory NetworkCallback{deliver} = do
-  revision <- getLastKnownRevision directory
   withGrpcContext "waitMessages" . forever $ do
     -- NOTE: We have not observed the watch (subscription) fail even when peers
     -- leave and we end up on a minority cluster.
     biDiStreaming conn (rpc @(Protobuf Watch "watch")) $ \send recv -> do
+      revision <- getLastKnownRevision directory
+      let startRevision = fromIntegral (revision + 1)
+      traceWith tracer WatchMessagesStartRevision{startRevision}
       -- NOTE: Request all keys starting with 'msg'. See also section KeyRanges
       -- in https://etcd.io/docs/v3.5/learning/api/#key-value-api
       let watchRequest =
@@ -384,34 +387,48 @@ waitMessages tracer conn directory NetworkCallback{deliver} = do
               & #rangeEnd .~ "msh" -- NOTE: g+1 to query prefixes
               & #startRevision .~ fromIntegral (revision + 1)
       send . NextElem $ defMessage & #createRequest .~ watchRequest
-      whileNext_ recv process
+      loop send recv
     -- Wait before re-trying
     threadDelay 1
  where
-  process res = do
-    let revision = fromIntegral $ res ^. #header . #revision
-    putLastKnownRevision directory revision
-    forM_ (res ^. #events) $ \event -> do
-      let value = event ^. #kv . #value
-      case decodeFull' value of
-        Left err ->
-          traceWith
-            tracer
-            FailedToDecodeValue
-              { key = decodeUtf8 $ event ^. #kv . #key
-              , value = encodeBase16 value
-              , reason = show err
-              }
-        Right msg -> deliver msg
+  loop send recv =
+    recv >>= \case
+      NoNextElem -> pure ()
+      NextElem res ->
+        if res ^. #canceled
+          then do
+            let compactRevision = res ^. #compactRevision
+            traceWith tracer WatchMessagesFallbackTo{compactRevision}
+            putLastKnownRevision directory . fromIntegral $ (compactRevision - 1) `max` 0
+            -- Gracefully close watch stream
+            send NoNextElem
+          else do
+            let revision = res ^. #header . #revision
+            putLastKnownRevision directory . fromIntegral $ revision `max` 0
+            forM_ (res ^. #events) process
+            loop send recv
+
+  process event = do
+    let value = event ^. #kv . #value
+    case decodeFull' value of
+      Left err ->
+        traceWith
+          tracer
+          FailedToDecodeValue
+            { key = decodeUtf8 $ event ^. #kv . #key
+            , value = encodeBase16 value
+            , reason = show err
+            }
+      Right msg -> deliver msg
 
 getLastKnownRevision :: MonadIO m => FilePath -> m Natural
 getLastKnownRevision directory = do
   liftIO $
     try (decodeFileStrict' $ directory </> "last-known-revision") >>= \case
       Right rev -> do
-        pure $ fromMaybe 1 rev
+        pure $ fromMaybe 0 rev
       Left (e :: IOException)
-        | isDoesNotExistError e -> pure 1
+        | isDoesNotExistError e -> pure 0
         | otherwise -> do
             fail $ "Failed to load last known revision: " <> show e
 
@@ -614,5 +631,7 @@ data EtcdLog
   | LowLeaseTTL {ttlRemaining :: DiffTime}
   | NoKeepAliveResponse
   | MatchingProtocolVersion {version :: ProtocolVersion}
+  | WatchMessagesStartRevision {startRevision :: Int64}
+  | WatchMessagesFallbackTo {compactRevision :: Int64}
   deriving stock (Eq, Show, Generic)
   deriving anyclass (ToJSON)
diff --git a/hydra-node/test/Hydra/NetworkSpec.hs b/hydra-node/test/Hydra/NetworkSpec.hs
@@ -23,11 +23,12 @@ import Hydra.Network (
   ProtocolVersion (..),
   WhichEtcd (..),
  )
-import Hydra.Network.Etcd (withEtcdNetwork)
+import Hydra.Network.Etcd (getClientPort, withEtcdNetwork)
 import Hydra.Network.Message (Message (..))
 import Hydra.Node.Network (NetworkConfiguration (..))
 import System.Directory (removeFile)
 import System.FilePath ((</>))
+import System.Process.Typed (runProcess_, shell)
 import Test.Aeson.GenericSpecs (Settings (..), defaultSettings, roundtripAndGoldenADTSpecsWithSettings)
 import Test.Hydra.Node.Fixture (alice, aliceSk, bob, bobSk, carol, carolSk)
 import Test.Network.Ports (randomUnusedTCPPorts, withFreePort)
@@ -177,14 +178,60 @@ spec = do
                 withEtcdNetwork @Int tracer v1 carolConfig recordCarol $ \_ -> do
                   broadcast n1 1001
                   waitCarol `shouldReturn` 1001
-                -- We can reset the last known view (internal implementation detail)
+
+      it "handles compaction and lost local state" $ \tracer -> do
+        withTempDir "test-etcd" $ \tmp -> do
+          failAfter 20 $ do
+            PeerConfig3{aliceConfig, bobConfig, carolConfig} <- setup3Peers tmp
+            (recordBob, waitBob, _) <- newRecordingCallback
+            (recordCarol, waitCarol, _) <- newRecordingCallback
+            withEtcdNetwork @Int tracer v1 aliceConfig noopCallback $ \n1 ->
+              withEtcdNetwork @Int tracer v1 bobConfig recordBob $ \_ -> do
+                -- First we send 5 messages with carol online
+                withEtcdNetwork @Int tracer v1 carolConfig recordCarol $ \_ -> do
+                  forM_ [1 .. 5] $ \msg -> do
+                    broadcast n1 msg
+                    waitBob `shouldReturn` msg
+                    waitCarol `shouldReturn` msg
+                -- Carol stopped and we continue sending messages
+                forM_ [5 .. 100] $ \msg -> do
+                  broadcast n1 msg
+                  waitBob `shouldReturn` msg
+                -- Even while carol is down, the etcd component would
+                -- "auto-compact" messages. By default down to 1000 messages
+                -- after/every 5 minutes. This is interesting as it should
+                -- result in carol never some messages, but is hard to test
+                -- (without waiting 5 minutes). Instead we issue a direct etcd
+                -- command to compact everything before revision 50.
+                runProcess_ . shell $
+                  "etcdctl compact 50 --endpoints=127.0.0.1:" <> show (getClientPort aliceConfig)
+                -- When carol starts now we would expect it to start catching up
+                -- from the earliest possible revision 50. While missing some
+                -- messages.
+                withEtcdNetwork @Int tracer v1 carolConfig recordCarol $ \_ -> do
+                  -- NOTE: Revision 50 may not correspond to message 50, so we
+                  -- only assert its some message bigger than 25 and expect to
+                  -- see all further messages to 100.
+                  firstMsg <- waitCarol
+                  firstMsg `shouldSatisfy` (> 25)
+                  forM_ [firstMsg + 1 .. 100] $ \msg ->
+                    waitCarol `shouldReturn` msg
+                  -- Carol should be able to receive new messages just fine.
+                  forM_ [101 .. 105] $ \msg -> do
+                    broadcast n1 msg
+                    waitCarol `shouldReturn` msg
+                -- Similarly, should carol lose its local state, we expect it to
+                -- see everything from the last compacted revision 50. We can
+                -- enforce this by removing the corresponding file (an internal
+                -- implementation detail)
                 removeFile (persistenceDir carolConfig </> "last-known-revision")
                 withEtcdNetwork @Int tracer v1 carolConfig recordCarol $ \_ -> do
-                  -- NOTE: The etcd component would "auto-compact" messages down
-                  -- to 1000 messages after 5 minutes. This would result in
-                  -- starting at 1001 here, but is hard to test (without waiting
-                  -- 5 minutes).
-                  forM_ messages $ \msg ->
+                  -- NOTE: Revision 50 may not correspond to message 50, so we
+                  -- only assert its some message bigger than 25 and expect to
+                  -- see all further messages to 105.
+                  firstMsg <- waitCarol
+                  firstMsg `shouldSatisfy` (> 25)
+                  forM_ [firstMsg + 1 .. 105] $ \msg -> do
                     waitCarol `shouldReturn` msg
 
       it "emits cluster id mismatch" $ \tracer -> do
diff --git a/nix/hydra/packages.nix b/nix/hydra/packages.nix
@@ -142,6 +142,7 @@ rec {
     buildInputs = [
       nativePkgs.hydra-node.components.tests.tests
       pkgs.check-jsonschema
+      pkgs.etcd # For etcdctl command in tests
     ];
   };
   hydra-cluster-tests = pkgs.mkShellNoCC {
