Fix blocking of newPersistentQueue (#2147)

See copilot summary (let's give this a spin)

---

* [x] CHANGELOG updated
* [x] Documentation update not needed
* [x] Haddocks update not needed
* [ ] No new TODOs introduced
- One FIXME which covers the still unknown part of why the queue was not
cleared

Author: noonio

hydra-node/hydra-node.cabal

@@ -87,6 +87,7 @@ library
     Hydra.Network
     Hydra.Network.Authenticate
     Hydra.Network.Etcd
+    Hydra.Network.EtcdBinary
     Hydra.Network.Message
     Hydra.NetworkVersions
     Hydra.Node
@@ -354,6 +355,7 @@ test-suite tests
     Hydra.OptionsSpec
     Hydra.PartySpec
     Hydra.PersistenceSpec
+    Hydra.PersistentQueueSpec
     Hydra.UtilsSpec
     Paths_hydra_node
     Spec

hydra-node/src/Hydra/Network.hs

@@ -53,6 +53,8 @@ data NetworkCallback msg m = NetworkCallback
 -- A `NetworkComponent` can have different inbound and outbound message types.
 type NetworkComponent m inbound outbound a = NetworkCallback inbound m -> (Network m outbound -> m a) -> m a
 
+-- * Types used by concrete implementations
+
 data WhichEtcd = EmbeddedEtcd | SystemEtcd
   deriving stock (Eq, Show, Generic)
   deriving anyclass (ToJSON, FromJSON)
@@ -61,8 +63,6 @@ instance Arbitrary WhichEtcd where
   shrink = genericShrink
   arbitrary = genericArbitrary
 
--- * Types used by concrete implementations
-
 -- | Configuration for a `Node` network layer.
 data NetworkConfiguration = NetworkConfiguration
   { persistenceDir :: FilePath

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

@@ -1,7 +1,5 @@
 {-# LANGUAGE OverloadedLabels #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TemplateHaskell #-}
-{-# OPTIONS_GHC -Wno-deferred-out-of-scope-variables #-}
 
 -- | Implements a Hydra network component using [etcd](https://etcd.io/).
 --
@@ -61,7 +59,6 @@ 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
 import Data.ByteString.Char8 qualified as BS8
 import Data.List ((\\))
@@ -77,9 +74,8 @@ import Hydra.Network (
   NetworkComponent,
   NetworkConfiguration (..),
   ProtocolVersion,
-  WhichEtcd (..),
  )
-import Hydra.Node.EmbedTH (embedExecutable)
+import Hydra.Network.EtcdBinary (getEtcdBinary)
 import Network.GRPC.Client (
   Address (..),
   CallParams (..),
@@ -108,9 +104,8 @@ import Network.GRPC.Etcd (
  )
 import System.Directory (createDirectoryIfMissing, listDirectory, removeFile)
 import System.Environment.Blank (getEnvironment)
-import System.FilePath (takeDirectory, (</>))
+import System.FilePath ((</>))
 import System.IO.Error (isDoesNotExistError)
-import System.Posix (ownerExecuteMode, ownerReadMode, ownerWriteMode, setFileMode)
 import System.Process (interruptProcessGroupOf)
 import System.Process.Typed (
   Process,
@@ -254,21 +249,6 @@ withEtcdNetwork tracer protocolVersion config callback action = do
 
   NetworkConfiguration{persistenceDir, listen, advertise, peers, whichEtcd} = config
 
--- | 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
-getEtcdBinary _ SystemEtcd = pure "etcd"
-getEtcdBinary persistenceDir EmbeddedEtcd =
-  let path = persistenceDir </> "bin" </> "etcd"
-   in installEtcd path >> pure path
-
--- | Install the embedded 'etcd' binary to given file path.
-installEtcd :: FilePath -> IO ()
-installEtcd fp = do
-  createDirectoryIfMissing True (takeDirectory fp)
-  BS.writeFile fp $(embedExecutable "etcd")
-  setFileMode fp (ownerReadMode .|. ownerWriteMode .|. ownerExecuteMode)
-
 -- | Check and write version on etcd cluster. This will retry until we are on a
 -- majority cluster and succeed. If the version does not match a corresponding
 -- 'Connectivity' message is sent via 'NetworkCallback'.
@@ -563,29 +543,31 @@ newPersistentQueue ::
   Natural ->
   m (PersistentQueue m a)
 newPersistentQueue path capacity = do
-  queue <- newTBQueueIO capacity
+  paths <- liftIO $ do
+    createDirectoryIfMissing True path
+    sort . mapMaybe readMaybe <$> listDirectory path
+  queue <- newTBQueueIO $ max (fromIntegral $ length paths) capacity
   highestId <-
-    try (loadExisting queue) >>= \case
+    try (loadExisting queue paths) >>= \case
       Left (_ :: IOException) -> do
+        -- XXX: This swallows and not logs the error
         liftIO $ createDirectoryIfMissing True path
         pure 0
       Right highest -> pure highest
   nextIx <- newTVarIO $ highestId + 1
   pure PersistentQueue{queue, nextIx, directory = path}
  where
-  loadExisting queue = do
-    paths <- liftIO $ listDirectory path
-    case sort $ mapMaybe readMaybe paths of
-      [] -> pure 0
-      idxs -> do
-        forM_ idxs $ \(idx :: Natural) -> do
-          bs <- readFileBS (path </> show idx)
-          case decodeFull' bs of
-            Left err ->
-              fail $ "Failed to decode item: " <> show err
-            Right item ->
-              atomically $ writeTBQueue queue (idx, item)
-        pure $ List.last idxs
+  loadExisting queue = \case
+    [] -> pure 0
+    idxs -> do
+      forM_ idxs $ \(idx :: Natural) -> do
+        bs <- readFileBS (path </> show idx)
+        case decodeFull' bs of
+          Left err ->
+            fail $ "Failed to decode item: " <> show err
+          Right item ->
+            atomically $ writeTBQueue queue (idx, item)
+      pure $ List.last idxs
 
 -- | Write a value to the queue, blocking if the queue is full.
 writePersistentQueue :: (ToCBOR a, MonadSTM m, MonadIO m) => PersistentQueue m a -> a -> m ()
@@ -595,6 +577,7 @@ writePersistentQueue PersistentQueue{queue, nextIx, directory} item = do
     modifyTVar' nextIx (+ 1)
     pure next
   writeFileBS (directory </> show next) $ serialize' item
+  -- XXX: We should trace when the queue is full
   atomically $ writeTBQueue queue (next, item)
 
 -- | Get the next value from the queue without removing it, blocking if the
@@ -610,6 +593,8 @@ popPersistentQueue PersistentQueue{queue, directory} item = do
   popped <- atomically $ do
     (ix, next) <- peekTBQueue queue
     if next == item
+      -- FIXME: why would we not call this? We saw the persistent queue reach
+      -- capacity and writing blocked while nothing seemed to clear it.
       then readTBQueue queue $> Just ix
       else pure Nothing
   case popped of

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

@@ -0,0 +1,29 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+-- | Embedding and installation of 'etcd' binary. In a dedicated module as HLS
+-- tends to choke on $(embedExecutable "etcd").
+module Hydra.Network.EtcdBinary where
+
+import Hydra.Prelude
+
+import Data.Bits ((.|.))
+import Hydra.Network (WhichEtcd (..))
+import Hydra.Node.EmbedTH (embedExecutable)
+import System.Directory (createDirectoryIfMissing)
+import System.FilePath (takeDirectory, (</>))
+import System.Posix (ownerExecuteMode, ownerReadMode, ownerWriteMode, setFileMode)
+
+-- | 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
+getEtcdBinary _ SystemEtcd = pure "etcd"
+getEtcdBinary persistenceDir EmbeddedEtcd =
+  let path = persistenceDir </> "bin" </> "etcd"
+   in installEtcd path >> pure path
+
+-- | Install the embedded 'etcd' binary to given file path.
+installEtcd :: FilePath -> IO ()
+installEtcd fp = do
+  createDirectoryIfMissing True (takeDirectory fp)
+  writeFileBS fp $(embedExecutable "etcd")
+  setFileMode fp (ownerReadMode .|. ownerWriteMode .|. ownerExecuteMode)

hydra-node/test/Hydra/PersistentQueueSpec.hs

@@ -0,0 +1,38 @@
+-- | Tests of the 'PersistentQueue'.
+module Hydra.PersistentQueueSpec where
+
+import Hydra.Prelude
+import Test.Hydra.Prelude
+
+import Hydra.Network.Etcd (newPersistentQueue, peekPersistentQueue, writePersistentQueue)
+import Test.QuickCheck (counterexample, generate, ioProperty)
+
+spec :: Spec
+spec = do
+  it "can be constructed" $ do
+    capacity <- generate arbitrary
+    withTempDir "persistent-queue" $ \dir -> do
+      void $ newPersistentQueue @_ @Int dir capacity
+
+  prop "is persistent with capacity" $ \(items :: [Int]) -> do
+    let capacity = fromIntegral $ length items
+    counterexample ("capacity: " <> show capacity) $
+      ioProperty $
+        withTempDir "persistent-queue" $ \dir -> do
+          q <- newPersistentQueue dir capacity
+          shouldNotBlock_ $ mapM (writePersistentQueue q) items
+          -- This is expected to block as we reached capacity
+          _ <- timeout 0.01 (writePersistentQueue q 123)
+          -- A new queue should be initialized with all the elements
+          q2 <- shouldNotBlock $ newPersistentQueue @_ @Int dir capacity
+          let expected = maybe 123 head (nonEmpty items)
+          peekPersistentQueue q2 `shouldReturn` expected
+
+shouldNotBlock :: HasCallStack => IO a -> IO a
+shouldNotBlock action = do
+  timeout 0.1 action >>= \case
+    Nothing -> failure "blocked unexpectedly"
+    Just a -> pure a
+
+shouldNotBlock_ :: HasCallStack => IO a -> IO ()
+shouldNotBlock_ = shouldNotBlock . void