Add cluster ID mismatch to connectivity messages

Signed-off-by: Sasha Bogicevic <[email protected]>

Author: v0d1ch

hydra-node/golden/ServerOutput/NetworkClusterIDMismatch.json

@@ -0,0 +1,10 @@
+{
+    "samples": [
+        {
+            "localClusterID": "",
+            "remotePeerClusterID": "",
+            "tag": "NetworkClusterIDMismatch"
+        }
+    ],
+    "seed": 1341180695
+}

hydra-node/golden/StateChanged/NetworkClusterIDMismatch.json

@@ -0,0 +1,15 @@
+{
+    "samples": [
+        {
+            "configuredParties": [
+                {
+                    "vkey": "d05f801e08648c87a1182768655803d4261ae2545573cd3a58b837f1541691de"
+                }
+            ],
+            "localClusterID": "",
+            "remotePeerClusterID": "",
+            "tag": "NetworkClusterIDMismatch"
+        }
+    ],
+    "seed": 1702753294
+}

hydra-node/json-schemas/api.yaml

@@ -80,6 +80,7 @@ channels:
           - $ref: "api.yaml#/components/messages/NetworkConnected"
           - $ref: "api.yaml#/components/messages/NetworkDisconnected"
           - $ref: "api.yaml#/components/messages/NetworkVersionMismatch"
+          - $ref: "api.yaml#/components/messages/NetworkClusterIDMismatch"
           - $ref: "api.yaml#/components/messages/HeadIsInitializing"
           - $ref: "api.yaml#/components/messages/Committed"
           - $ref: "api.yaml#/components/messages/HeadIsOpen"
@@ -487,6 +488,13 @@ components:
       payload:
         $ref: "api.yaml#/components/schemas/NetworkVersionMismatch"
 
+    NetworkClusterIDMismatch:
+      title: NetworkClusterIDMismatch
+      description: |
+        The configured peers do not match with other nodes on the network.
+      payload:
+        $ref: "api.yaml#/components/schemas/NetworkClusterIDMismatch"
+
     HeadIsInitializing:
       title: HeadIsInitializing
       description: |
@@ -732,6 +740,7 @@ components:
         - $ref: "api.yaml#/components/schemas/NetworkConnected"
         - $ref: "api.yaml#/components/schemas/NetworkDisconnected"
         - $ref: "api.yaml#/components/schemas/NetworkVersionMismatch"
+        - $ref: "api.yaml#/components/schemas/NetworkClusterIDMismatch"
         - $ref: "api.yaml#/components/schemas/PeerConnected"
         - $ref: "api.yaml#/components/schemas/PeerDisconnected"
         - $ref: "api.yaml#/components/schemas/HeadIsInitializing"
@@ -985,6 +994,29 @@ components:
         timestamp:
           $ref: "api.yaml#/components/schemas/UTCTime"
 
+    NetworkClusterIDMismatch:
+      type: object
+      required:
+        - tag
+        - seq
+        - timestamp
+      properties:
+        tag:
+          type: string
+          enum: ["NetworkClusterIDMismatch"]
+        localClusterID:
+          type: string
+        remotePeerClusterID:
+          type: string
+        configuredParties:
+          type: array
+          items:
+            $ref: "api.yaml#/components/schemas/Party"
+        seq:
+          $ref: "api.yaml#/components/schemas/SequenceNumber"
+        timestamp:
+          $ref: "api.yaml#/components/schemas/UTCTime"
+
     HeadIsInitializing:
       type: object
       required:

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

@@ -249,6 +249,7 @@ mkTimedServerOutputFromStateEvent event =
     StateChanged.NetworkConnected -> Just NetworkConnected
     StateChanged.NetworkDisconnected -> Just NetworkDisconnected
     StateChanged.NetworkVersionMismatch{..} -> Just NetworkVersionMismatch{..}
+    StateChanged.NetworkClusterIDMismatch{..} -> Just NetworkClusterIDMismatch{..}
     StateChanged.PeerConnected{..} -> Just PeerConnected{..}
     StateChanged.PeerDisconnected{..} -> Just PeerDisconnected{..}
     StateChanged.TransactionReceived{} -> Nothing

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

@@ -142,6 +142,10 @@ data ServerOutput tx
       { ourVersion :: ProtocolVersion
       , theirVersion :: Maybe ProtocolVersion
       }
+  | NetworkClusterIDMismatch
+      { localClusterID :: Text
+      , remotePeerClusterID :: Text
+      }
   | PeerConnected {peer :: Host}
   | PeerDisconnected {peer :: Host}
   | HeadIsInitializing {headId :: HeadId, parties :: [Party]}
@@ -273,6 +277,7 @@ prepareServerOutput config response =
     NetworkConnected -> encodedResponse
     NetworkDisconnected -> encodedResponse
     NetworkVersionMismatch{} -> encodedResponse
+    NetworkClusterIDMismatch{} -> encodedResponse
     PeerConnected{} -> encodedResponse
     PeerDisconnected{} -> encodedResponse
     SnapshotSideLoaded{} -> encodedResponse

hydra-node/src/Hydra/HeadLogic.hs

@@ -103,14 +103,17 @@ import Hydra.Tx.OnChainId (OnChainId)
 import Hydra.Tx.Party (Party (vkey))
 import Hydra.Tx.Snapshot (ConfirmedSnapshot (..), Snapshot (..), SnapshotNumber, SnapshotVersion, getSnapshot)
 
-onConnectionEvent :: Network.Connectivity -> Outcome tx
-onConnectionEvent = \case
+
+onConnectionEvent :: [Party] -> Network.Connectivity -> Outcome tx
+onConnectionEvent configuredParties = \case
   Network.NetworkConnected ->
     newState NetworkConnected
   Network.NetworkDisconnected ->
     newState NetworkDisconnected
   Network.VersionMismatch{ourVersion, theirVersion} ->
     newState NetworkVersionMismatch{ourVersion, theirVersion}
+  Network.ClusterIDMismatch{localClusterID, remotePeerClusterID} ->
+    newState NetworkClusterIDMismatch{localClusterID, remotePeerClusterID, configuredParties}
   Network.PeerConnected{peer} ->
     newState PeerConnected{peer}
   Network.PeerDisconnected{peer} ->
@@ -1313,7 +1316,7 @@ update ::
   Outcome tx
 update env ledger st ev = case (st, ev) of
   (_, NetworkInput _ (ConnectivityEvent conn)) ->
-    onConnectionEvent conn
+    onConnectionEvent env.otherParties conn
   (Idle _, ClientInput Init) ->
     onIdleClientInit env
   (Idle _, ChainInput Observation{observedTx = OnInitTx{headId, headSeed, headParameters, participants}, newChainState}) ->
@@ -1426,6 +1429,7 @@ aggregate st = \case
   NetworkConnected -> st
   NetworkDisconnected -> st
   NetworkVersionMismatch{} -> st
+  NetworkClusterIDMismatch{} -> st
   PeerConnected{} -> st
   PeerDisconnected{} -> st
   HeadInitialized{parameters = parameters@HeadParameters{parties}, headId, headSeed, chainState} ->
@@ -1793,6 +1797,7 @@ aggregateChainStateHistory history = \case
   NetworkConnected -> history
   NetworkDisconnected -> history
   NetworkVersionMismatch{} -> history
+  NetworkClusterIDMismatch{} -> history
   PeerConnected{} -> history
   PeerDisconnected{} -> history
   HeadInitialized{chainState} -> pushNewState chainState history

hydra-node/src/Hydra/HeadLogic/Outcome.hs

@@ -60,6 +60,11 @@ data StateChanged tx
       { ourVersion :: ProtocolVersion
       , theirVersion :: Maybe ProtocolVersion
       }
+  | NetworkClusterIDMismatch
+      { localClusterID :: Text
+      , remotePeerClusterID :: Text
+      , configuredParties :: [Party]
+      }
   | HeadInitialized
       { parameters :: HeadParameters
       , chainState :: ChainStateType tx

hydra-node/src/Hydra/Network.hs

@@ -185,6 +185,10 @@ data Connectivity
       { ourVersion :: ProtocolVersion
       , theirVersion :: Maybe ProtocolVersion
       }
+  | ClusterIDMismatch
+      { localClusterID :: Text
+      , remotePeerClusterID :: Text
+      }
   deriving stock (Generic, Eq, Show)
   deriving anyclass (ToJSON)
 

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

@@ -56,9 +56,10 @@ import Control.Concurrent.Class.MonadSTM (
   writeTVar,
  )
 import Control.Exception (IOException)
-import Control.Lens ((^.), (^..))
+import Control.Lens ((^.), (^..), (^?))
 import Data.Aeson (decodeFileStrict', encodeFile)
 import Data.Aeson qualified as Aeson
+import Data.Aeson.Lens qualified as Aeson
 import Data.Aeson.Types (Value)
 import Data.Bits ((.|.))
 import Data.ByteString qualified as BS
@@ -138,12 +139,12 @@ withEtcdNetwork tracer protocolVersion config callback action = do
   envVars <- Map.fromList <$> getEnvironment
   withProcessInterrupt (etcdCmd etcdBinPath envVars) $ \p -> do
     race_ (waitExitCode p >>= \ec -> fail $ "Sub-process etcd exited with: " <> show ec) $ do
-      race_ (traceStderr p) $ do
+      race_ (traceStderr p callback) $ do
         -- XXX: cleanup reconnecting through policy if other threads fail
         doneVar <- newTVarIO False
         -- NOTE: The connection to the server is set up asynchronously; the
         -- first rpc call will block until the connection has been established.
-        withConnection (connParams doneVar) grpcServer $ \conn ->
+        withConnection (connParams doneVar) grpcServer $ \conn -> do
           -- REVIEW: checkVersion blocks if used on main thread - why?
           withAsync (checkVersion tracer conn protocolVersion callback) $ \_ -> do
             race_ (pollConnectivity tracer conn advertise callback) $
@@ -188,12 +189,22 @@ withEtcdNetwork tracer protocolVersion config callback action = do
   -- be used by default still.
   clientPort = 2379 + port listen - 5001
 
-  traceStderr p =
+  traceStderr p NetworkCallback{onConnectivity} =
     forever $ do
       bs <- BS.hGetLine (getStderr p)
       case Aeson.eitherDecodeStrict bs of
         Left err -> traceWith tracer FailedToDecodeLog{log = decodeUtf8 bs, reason = show err}
-        Right v -> traceWith tracer $ EtcdLog{etcd = v}
+        Right v -> do
+          let expectedClusterMismatch = do
+                level' <- bs ^? Aeson.key "level" . Aeson.nonNull
+                msg' <- bs ^? Aeson.key "msg" . Aeson.nonNull
+                localClusterId <- bs ^? Aeson.key "local-member-cluster-id" . Aeson.nonNull
+                remoteClusterId <- bs ^? Aeson.key "remote-peer-cluster-id" . Aeson.nonNull
+                pure (level', msg', localClusterId, remoteClusterId)
+          case expectedClusterMismatch of
+            Just (Aeson.String "error", Aeson.String "request sent was ignored due to cluster ID mismatch", Aeson.String localClusterID, Aeson.String remotePeerClusterID) ->
+              onConnectivity ClusterIDMismatch{localClusterID, remotePeerClusterID}
+            _ -> traceWith tracer $ EtcdLog{etcd = v}
 
   -- XXX: Could use TLS to secure peer connections
   -- XXX: Could use discovery to simplify configuration

hydra-node/test/Hydra/HeadLogicSpec.hs

@@ -657,6 +657,7 @@ spec =
                       PeerConnected{} -> True
                       PeerDisconnected{} -> True
                       NetworkVersionMismatch{} -> True
+                      NetworkClusterIDMismatch{} -> True
                       NetworkConnected{} -> True
                       NetworkDisconnected{} -> True
                       _ -> False