cachix 1.3 → 1.3.1

raw patch · 4 files changed

+51/−28 lines, 4 files

Files

• CHANGELOG.md +6/−0
• cachix.cabal +1/−1
• src/Cachix/Client/Commands.hs +20/−15
• src/Cachix/Client/PushQueue.hs +24/−12

@@ -5,6 +5,12 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [1.3.1] - 2023-03-09++### Fixed++- Signal handling in watch-exec & watch-store commands.+ ## [1.3] - 2023-03-06  ### Added

@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.3+version:            1.3.1 license:            Apache-2.0 license-file:       LICENSE copyright:          2018 Domen Kozar

@@ -56,6 +56,7 @@ import Servant.Conduit () import System.Directory (doesFileExist) import System.IO (hIsTerminalDevice)+import qualified System.Posix.Signals as Signals import qualified System.Process  -- TODO: check that token actually authenticates!@@ -201,21 +202,25 @@         hDuplicateTo stderr stdout -- redirect all stdout to stderr         WatchStore.startWorkers (pushParamsStore pushParams) (numJobs pushOpts) pushParams -  Async.withAsync watch $ \watchThread ->-    bracketOnError-      (getProcessHandle <$> System.Process.createProcess process)-      ( \processHandle -> do-          -- Terminate the process-          uninterruptibleMask_ (System.Process.terminateProcess processHandle)-          -- Wait for the process to clean up and exit-          _ <- System.Process.waitForProcess processHandle-          -- Stop watching the store and wait for all paths to be pushed-          Async.cancel watchThread-      )-      $ \processHandle -> do-        exitCode <- System.Process.waitForProcess processHandle-        Async.cancel watchThread-        exitWith exitCode+  Async.withAsync watch $ \watchThread -> do+    -- Throw any errors encountered by the workers+    Async.link watchThread++    exitCode <-+      bracketOnError+        (getProcessHandle <$> System.Process.createProcess process)+        ( \processHandle -> do+            -- Terminate the process+            uninterruptibleMask_ (System.Process.terminateProcess processHandle)+            -- Wait for the process to clean up and exit+            _ <- System.Process.waitForProcess processHandle+            -- Stop watching the store and wait for all paths to be pushed+            Signals.raiseSignal Signals.sigINT+        )+        System.Process.waitForProcess++    Signals.raiseSignal Signals.sigINT+    exitWith exitCode   where     getProcessHandle (_, _, _, processHandle) = processHandle 

@@ -16,12 +16,14 @@ import qualified Cachix.Client.Push as Push import Cachix.Client.Retry (retryAll) import Control.Concurrent.Async+import Control.Concurrent.Extra (once) import Control.Concurrent.STM (TVar, modifyTVar', newTVarIO, readTVar) import qualified Control.Concurrent.STM.Lock as Lock import qualified Control.Concurrent.STM.TBQueue as TBQueue import qualified Data.Set as S import Hercules.CNix.Store (StorePath) import Protolude+import qualified System.Posix.Signals as Signals import qualified System.Systemd.Daemon as Systemd  type Queue = TBQueue.TBQueue StorePath@@ -60,15 +62,21 @@       (,,) <$> TBQueue.newTBQueue 10000 <*> TBQueue.newTBQueue 10000 <*> Lock.new   let queryWorkerState = QueryWorkerState newQueryQueue S.empty newLock   queryWorker <- async $ queryLoop queryWorkerState newPushQueue pushParams+   -- start push workers   stopProducerCallback <- mkProducer newQueryQueue   progress <- newTVarIO 0   let pushWorkerState = PushWorkerState newPushQueue progress   pushWorker <- async $ replicateConcurrently_ numWorkers $ worker pushParams pushWorkerState-  let drainQueue =-        exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState -  (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker] `finally` drainQueue+  -- Gracefully shutdown the workers on interrupt+  let handler =+        Signals.CatchOnce $+          exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState+  for_ [Signals.sigINT, Signals.sigTERM] $ \signal ->+    Signals.installHandler signal handler Nothing++  (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker]   case eitherException of     Left exc | fromException exc == Just StopWorker -> return ()     Left exc -> throwIO exc@@ -100,17 +108,21 @@  -- | Stop watching the store and push all pending store paths. exitOnceQueueIsEmpty :: IO () -> Async () -> Async () -> QueryWorkerState -> PushWorkerState -> IO ()-exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = do-  putTextError "Stopped watching /nix/store and waiting for queue to empty ..."+exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState =+  join . once $ do+    putTextError "Stopped watching /nix/store and waiting for queue to empty ..." -  -- Skip uploading the remaining paths when run in an interruptible mask.-  getMaskingState >>= \case-    MaskedUninterruptible -> stopWorkers-    _ -> do-      void Systemd.notifyStopping-      stopProducerCallback-      go+    -- Skip uploading the remaining paths when run in an interruptible mask to avoid hanging on IO.+    getMaskingState >>= \case+      MaskedUninterruptible -> stopWorkers+      _ -> do+        void Systemd.notifyStopping+        stopProducerCallback+        go   where+    -- We can safely skip calling the interrupt handler on Nix and+    -- avoid seeing the generic interrupt message.+    -- Nix only uses it to cancel file transfers, which we don't use.     stopWorkers = do       cancelWith queryWorker StopWorker       cancelWith pushWorker StopWorker