Low lease ttl (#2148)

When we detect a lease has expired, just re-request it, instead of trying to use it indefinitely.

---

* [x] CHANGELOG updated
* [x] Documentation update not needed
* [x] Haddocks update not needed
* [x] No new TODOs introduced

Author: noonio

hydra-node/hydra-node.cabal

@@ -421,6 +421,7 @@ test-suite tests
     , text
     , time
     , tls
+    , typed-process
     , websockets
 
   build-tool-depends: hspec-discover:hspec-discover

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

@@ -450,23 +450,27 @@ pollConnectivity tracer conn advertise NetworkCallback{onConnectivity} = do
       -- Write our alive key using lease
       writeAlive leaseId
       traceWith tracer CreatedLease{leaseId}
-      withKeepAlive leaseId $ \keepAlive ->
-        forever $ do
-          -- Keep our lease alive
-          ttlRemaining <- keepAlive
-          when (ttlRemaining < 1) $
-            traceWith tracer LowLeaseTTL{ttlRemaining}
-          -- Determine alive peers
-          alive <- getAlive
-          let othersAlive = alive \\ [advertise]
-          seenAlive <- atomically $ swapTVar seenAliveVar othersAlive
-          forM_ (othersAlive \\ seenAlive) $ onConnectivity . PeerConnected
-          forM_ (seenAlive \\ othersAlive) $ onConnectivity . PeerDisconnected
-          -- Wait roughly ttl / 2
-          threadDelay (ttlRemaining / 2)
+      withKeepAlive leaseId (aliveLoop seenAliveVar)
  where
+  aliveLoop seenAliveVar keepAlive = do
+    -- Keep our lease alive
+    ttlRemaining <- keepAlive
+    if ttlRemaining <= 0
+      then
+        -- The keep alive did not work as no time to live remaining. Get a new lease instead
+        traceWith tracer LowLeaseTTL{ttlRemaining}
+      else do
+        -- Determine alive peers
+        alive <- getAlive
+        let othersAlive = alive \\ [advertise]
+        seenAlive <- atomically $ swapTVar seenAliveVar othersAlive
+        forM_ (othersAlive \\ seenAlive) $ onConnectivity . PeerConnected
+        forM_ (seenAlive \\ othersAlive) $ onConnectivity . PeerDisconnected
+        threadDelay 1
+        aliveLoop seenAliveVar keepAlive
+
   onGrpcException seenAliveVar GrpcException{grpcError}
-    | grpcError == GrpcUnavailable || grpcError == GrpcDeadlineExceeded = do
+    | grpcError `elem` [GrpcUnavailable, GrpcDeadlineExceeded, GrpcCancelled] = do
         onConnectivity NetworkDisconnected
         atomically $ writeTVar seenAliveVar []
         threadDelay 1
@@ -483,7 +487,7 @@ pollConnectivity tracer conn advertise NetworkCallback{onConnectivity} = do
       void . action $ do
         send $ NextElem $ defMessage & #id .~ leaseId
         recv >>= \case
-          NextElem res -> pure . fromIntegral $ res ^. #ttl
+          NextElem res -> pure $ res ^. #ttl
           NoNextElem -> do
             traceWith tracer NoKeepAliveResponse
             pure 0
@@ -622,7 +626,7 @@ data EtcdLog
   | FailedToDecodeLog {log :: Text, reason :: Text}
   | FailedToDecodeValue {key :: Text, value :: Text, reason :: Text}
   | CreatedLease {leaseId :: Int64}
-  | LowLeaseTTL {ttlRemaining :: DiffTime}
+  | LowLeaseTTL {ttlRemaining :: Int64}
   | NoKeepAliveResponse
   | MatchingProtocolVersion {version :: ProtocolVersion}
   deriving stock (Eq, Show, Generic)

hydra-node/test/Hydra/NetworkSpec.hs

@@ -28,6 +28,7 @@ import Hydra.Network.Message (Message (..))
 import Hydra.Node.Network (NetworkConfiguration (..))
 import System.Directory (removeFile)
 import System.FilePath ((</>))
+import System.Process.Typed (readProcessStdout_, 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)
@@ -139,6 +140,30 @@ spec = do
                 waitFor NetworkConnected
                 waitFor $ PeerConnected carolConfig.advertise
 
+      it "handles expired lease" $ \tracer -> do
+        withTempDir "test-etcd" $ \tmp -> do
+          failAfter 5 $ do
+            PeerConfig2{aliceConfig, bobConfig} <- setup2Peers tmp
+            -- Record and assert connectivity events from alice's perspective
+            (recordReceived, _, waitConnectivity) <- newRecordingCallback
+            let
+              waitFor :: HasCallStack => Connectivity -> IO ()
+              waitFor = waitEq waitConnectivity 60
+            withEtcdNetwork @Int tracer v1 aliceConfig recordReceived $ \_ -> do
+              withEtcdNetwork @Int tracer v1 bobConfig noopCallback $ \_ -> do
+                waitFor NetworkConnected
+                waitFor $ PeerConnected bobConfig.advertise
+                -- Expire all leases manually to simulate a keepAlive coming too
+                -- late. Note that we do not distinguish which is which so
+                -- alice's lease will also be killed, but does not matter here.
+                let endpoints = "--endpoints=" <> show (listen aliceConfig)
+                output <- readProcessStdout_ . shell $ "etcdctl lease list " <> endpoints
+                let leases = drop 1 $ lines $ decodeUtf8 output
+                forM_ leases $ \lease ->
+                  runProcess_ . shell $ "etcdctl lease revoke " <> endpoints <> " " <> toString lease
+                -- Alice sees bob disconnected and connected again
+                waitFor $ PeerConnected bobConfig.advertise
+
       it "checks protocol version" $ \tracer -> do
         withTempDir "test-etcd" $ \tmp -> do
           failAfter 10 $ do