packages feed

cachix 1.12.0 → 1.12.1

raw patch · 10 files changed

+416/−52 lines, 10 files

Files

CHANGELOG.md view
@@ -5,6 +5,17 @@ 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). +## [Unreleased]++## [1.12.1] - 2026-08-31++### Fixed++- auth: explain that generating signing keys requires a personal administrator token+- deploy: time out stalled secure connection setup+- deploy: preserve trusted public keys for existing substituters when downloading store paths+- daemon: notify waiting clients when push jobs fail and prevent events after completion+ ## [1.12.0] - 2026-08-10  ### Added
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.12.0+version:            1.12.1 synopsis:   Command-line client for Nix binary cache hosting https://cachix.org @@ -249,6 +249,7 @@   main-is:            Main.hs   hs-source-dirs:     test   other-modules:+    CommandConfigSpec     Daemon.NarinfoQuerySpec     Daemon.PostBuildHookSpec     Daemon.ProtocolSpec
src/Cachix/Client/Command/Config.hs view
@@ -7,6 +7,7 @@ import Cachix.API.Error import Cachix.Client.Config qualified as Config import Cachix.Client.Env (Env (..))+import Cachix.Client.Exception (CachixException (AccessDeniedBinaryCache)) import Cachix.Client.Retry (retryClientM) import Cachix.Client.Secrets   ( SigningKey (SigningKey),@@ -19,6 +20,7 @@ import Data.String.Here import Data.Text qualified as T import Data.Text.IO qualified as T.IO+import Network.HTTP.Types.Status (status403) import Protolude hiding (toS) import Protolude.Conv import Servant.API (NoContent (..))@@ -40,9 +42,12 @@       signingKeyCreate = SigningKeyCreate.SigningKeyCreate (toS $ B64.encode pk)       bcc = Config.BinaryCacheConfig name signingKey   -- we first validate if key can be added to the binary cache+  createKeyResult <- retryClientM (clientenv env) (API.createKey cachixClient authToken name signingKeyCreate)   (_ :: NoContent) <--    escalate-      =<< retryClientM (clientenv env) (API.createKey cachixClient authToken name signingKeyCreate)+    case createKeyResult of+      Left err+        | isErr err status403 -> throwIO $ generateKeypairAccessDenied name+      _ -> escalate createKeyResult   -- if key was successfully added, write it to the config   -- TODO: warn if binary cache with the same key already exists   let cfg = config env & Config.setBinaryCaches [bcc]@@ -67,3 +72,12 @@   |] ::         Text     )++generateKeypairAccessDenied :: Text -> CachixException+generateKeypairAccessDenied name =+  AccessDeniedBinaryCache+    [iTrim|+Cannot create a signing key for binary cache ${name}.++This operation requires a personal auth token belonging to a Cachix account that administers the cache. Per-cache read/write tokens can push and pull, but they can't manage signing keys.+    |]
src/Cachix/Daemon/PushManager.hs view
@@ -20,6 +20,7 @@     -- * Query     filterPushJobs,     getFailedPushJobs,+    failPushJob,     failPendingJobs,      -- * Store paths@@ -65,6 +66,7 @@ import Cachix.Types.BinaryCache qualified as BinaryCache import Conduit qualified as C import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar qualified as MVar import Control.Concurrent.STM.TVar import Control.Monad.Catch qualified as E import Control.Monad.IO.Unlift (MonadUnliftIO)@@ -93,6 +95,7 @@   pmPushJobs <- newTVarIO mempty   pmPendingJobCount <- newTVarIO 0   pmStorePathIndex <- newTVarIO mempty+  pmJobLock <- MVar.newMVar ()   pmTaskQueue <- atomically newTaskQueue   pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)   pmLastEventTimestamp <- newTVarIO =<< getCurrentTime@@ -211,37 +214,42 @@     handleMissingPushJob =       Katip.logLocM Katip.ErrorS $ Katip.ls $ "Push job " <> (show pushId :: Text) <> " not found" --- | Apply an update to many push jobs atomically. After the update, a job--- transitions to a terminal state if either the update made it processed--- directly or it leaves 'pushQueue' empty. Returns the jobs that transitioned--- to a terminal state in this call.+-- | Apply an update to many push jobs atomically. Jobs that are already in a+-- terminal state are left untouched. After the update, a job transitions to+-- a terminal state if either the update made it processed directly or it+-- leaves 'pushQueue' empty. Jobs that reach a terminal state are removed from+-- the store path index so that in-flight uploads stop reporting to them.+-- Returns the jobs that were updated; use 'PushJob.isProcessed' to pick out+-- the ones that finished in this call. ----- Mark, completion check, and pending-counter decrement happen in one STM--- transaction so concurrent workers cannot race past the empty-queue check.+-- Mark, completion check, index cleanup, and pending-counter decrement happen+-- in one STM transaction so concurrent workers cannot race past the+-- empty-queue check. applyPushJobUpdates ::   (Foldable t) =>   t Protocol.PushRequestId ->   (UTCTime -> PushJob -> PushJob) ->   PushManager [PushJob] applyPushJobUpdates pushIds update = do-  PushManagerEnv {pmPushJobs, pmPendingJobCount} <- ask+  PushManagerEnv {pmPushJobs, pmPendingJobCount, pmStorePathIndex} <- ask   ts <- liftIO getCurrentTime   liftIO $ atomically $ do     jobs <- readTVar pmPushJobs-    let (jobs', finished) = foldl' (step ts) (jobs, []) pushIds+    let (jobs', updated) = foldl' (step ts) (jobs, []) pushIds+        finished = filter PushJob.isProcessed updated     writeTVar pmPushJobs jobs'     modifyTVar' pmPendingJobCount (subtract (length finished))-    pure finished+    unless (null finished) $ do+      let finishedIds = Set.fromList (map PushJob.pushId finished)+      modifyTVar' pmStorePathIndex $ HashMap.map (Seq.filter (`Set.notMember` finishedIds))+    pure updated   where     step ts (!jobs, acc) pushId =       case HashMap.lookup pushId jobs of         Just job           | not (PushJob.isProcessed job) ->               let job' = transitionIfDone ts (update ts job)-                  jobs' = HashMap.insert pushId job' jobs-               in if PushJob.isProcessed job'-                    then (jobs', job' : acc)-                    else (jobs', acc)+               in (HashMap.insert pushId job' jobs, job' : acc)         _ -> (jobs, acc)      transitionIfDone ts job@@ -252,28 +260,43 @@             else PushJob.complete ts job       | otherwise = job -failPushJob :: Protocol.PushRequestId -> PushManager ()-failPushJob pushId = void $ applyPushJobUpdates [pushId] PushJob.fail+failPushJob :: Protocol.PushRequestId -> Text -> PushManager ()+failPushJob pushId reason = void $ failPushJobs [pushId] reason --- | Mark every non-terminal job as failed and emit failure events for any--- paths still in their queues. Failed jobs stay in 'pmPushJobs' so a later--- 'getFailedPushJobs' call sees them and the daemon exits with the right--- code; the natural path removes jobs via 'pushFinished'.+-- | Mark every non-terminal job as failed and emit failure events for their+-- remaining paths. failPendingJobs :: Text -> PushManager [PushJob] failPendingJobs reason = do   pmPushJobs <- asks pmPushJobs   allIds <- HashMap.keys <$> liftIO (readTVarIO pmPushJobs)-  failed <- applyPushJobUpdates allIds PushJob.fail+  failPushJobs allIds reason +-- | Mark the given jobs as failed, emitting a 'PushStorePathFailed' event for+-- each path that will no longer be pushed, followed by 'PushFinished' so that+-- subscribers stop waiting. The jobs are dropped from the store path index at+-- the same time, so uploads that are still in flight do not emit any further+-- events for them. Failed jobs stay in 'pmPushJobs' so a later+-- 'getFailedPushJobs' call sees them and the daemon exits with the right+-- code; the natural path removes jobs via 'pushFinished'.+failPushJobs :: (Foldable t) => t Protocol.PushRequestId -> Text -> PushManager [PushJob]+failPushJobs pushIds reason = withJobLock $ do+  failedJobs <- applyPushJobUpdates pushIds PushJob.fail+   ts <- liftIO getCurrentTime   sendPushEvent <- asks pmOnPushEvent-  for_ failed $ \job -> do+  for_ failedJobs $ \job -> do     let pid = PushJob.pushId job-    for_ (PushJob.pushQueue job) $ \path ->+    for_ (unpushedPaths job) $ \path ->       sendStorePathEventAt ts [pid] (PushStorePathFailed path reason)     liftIO $ sendPushEvent pid (PushEvent ts pid PushFinished) -  pure failed+  pure failedJobs+  where+    -- A job that fails before closure resolution has an empty queue; fall+    -- back to the requested paths so subscribers still see the failure.+    unpushedPaths job+      | isNothing (PushJob.startedAt job) = Set.fromList $ Protocol.storePaths (pushRequest job)+      | otherwise = PushJob.pushQueue job  pendingJobCount :: PushManager Int pendingJobCount = do@@ -316,20 +339,27 @@     countQueuedPaths acc job = acc + fromIntegral (Set.size $ pushQueue job)  resolvePushJob :: Protocol.PushRequestId -> PushJob.ResolvedClosure FilePath -> PushManager ()-resolvePushJob pushId closure = do+resolvePushJob pushId closure = withJobLock $ do   Katip.logLocM Katip.DebugS $ Katip.ls $ showClosureStats closure -  finishedJobs <- applyPushJobUpdates [pushId] (PushJob.populateQueue closure)+  updatedJobs <- applyPushJobUpdates [pushId] (PushJob.populateQueue closure) -  withPushJob pushId $ \pushJob -> do+  -- A job that already reached a terminal state (e.g. failed during shutdown)+  -- must not start pushing, or it would emit events after 'PushFinished'.+  when (null updatedJobs) $+    Katip.logLocM Katip.DebugS $+      Katip.ls $+        "Push job " <> (show pushId :: Text) <> " is no longer active, ignoring resolved closure"++  for_ updatedJobs $ \pushJob -> do     pushStarted pushJob     let skippedPaths = Set.difference (PushJob.rcAllPaths closure) (PushJob.rcMissingPaths closure)     ts <- liftIO getCurrentTime     forM_ skippedPaths $ \path ->       sendStorePathEventAt ts [pushId] (PushStorePathSkipped path)     queueStorePaths pushId $ Set.toList (PushJob.rcMissingPaths closure)--  for_ finishedJobs pushFinished+    when (PushJob.isProcessed pushJob) $+      pushFinished pushJob   where     showClosureStats :: PushJob.ResolvedClosure FilePath -> Text     showClosureStats PushJob.ResolvedClosure {..} =@@ -362,7 +392,7 @@   where     failJob :: SomeException -> PushManager ()     failJob err = do-      failPushJob pushId+      failPushJob pushId $ "Failed to resolve closure: " <> toS (displayException err)        Katip.katipAddContext (Katip.sl "error" (displayException err)) $         Katip.logLocM Katip.ErrorS $@@ -385,7 +415,7 @@         -- Emit PushStorePathInvalid events for invalid paths         ts <- liftIO getCurrentTime         forM_ errors $ \(path, err) ->-          sendStorePathEventAt ts [pushId] (PushStorePathInvalid path (formatStorePathError err))+          sendStorePathEventForActiveJobsAt ts [pushId] (PushStorePathInvalid path (formatStorePathError err))          paths <- computeClosure store validPaths @@ -399,7 +429,7 @@   where     failJob :: SomeException -> PushManager ()     failJob err = do-      failPushJob pushId+      failPushJob pushId $ "Failed to query missing paths: " <> toS (displayException err)        Katip.katipAddContext (Katip.sl "error" (displayException err)) $         Katip.logLocM Katip.ErrorS $@@ -567,33 +597,51 @@     sendPushEvent pushId (PushEvent timestamp pushId msg)  pushStorePathAttempt :: FilePath -> Int64 -> RetryStatus -> PushManager ()-pushStorePathAttempt storePath size retryStatus = do+pushStorePathAttempt storePath size retryStatus = withJobLock $ do   let pushRetryStatus = newPushRetryStatus retryStatus   pushIds <- lookupStorePathIndex storePath   sendStorePathEvent pushIds (PushStorePathAttempt storePath size pushRetryStatus)  pushStorePathProgress :: FilePath -> Int64 -> Int64 -> PushManager ()-pushStorePathProgress storePath currentBytes newBytes = do+pushStorePathProgress storePath currentBytes newBytes = withJobLock $ do   pushIds <- lookupStorePathIndex storePath   sendStorePathEvent pushIds (PushStorePathProgress storePath currentBytes newBytes)  pushStorePathDone :: FilePath -> PushManager ()-pushStorePathDone storePath = do-  pushIds <- lookupStorePathIndex storePath-  finishedJobs <- applyPushJobUpdates pushIds (\_ -> PushJob.markStorePathPushed storePath)-  sendStorePathEvent pushIds (PushStorePathDone storePath)-  for_ finishedJobs pushFinished-  removeStorePath storePath+pushStorePathDone storePath =+  finishStorePath storePath PushJob.markStorePathPushed (PushStorePathDone storePath)  pushStorePathFailed :: FilePath -> Text -> PushManager ()-pushStorePathFailed storePath errMsg = do+pushStorePathFailed storePath errMsg =+  finishStorePath storePath PushJob.markStorePathFailed (PushStorePathFailed storePath errMsg)++-- | Record the outcome of a store path push on every job that is still+-- waiting for it, then emit the event to those jobs only. Jobs that already+-- reached a terminal state (e.g. failed during shutdown) receive nothing, so+-- 'PushFinished' stays the last event a subscriber sees.+finishStorePath :: FilePath -> (FilePath -> PushJob -> PushJob) -> PushEventMessage -> PushManager ()+finishStorePath storePath markStorePath msg = withJobLock $ do   pushIds <- lookupStorePathIndex storePath-  finishedJobs <- applyPushJobUpdates pushIds (\_ -> PushJob.markStorePathFailed storePath)-  sendStorePathEvent pushIds (PushStorePathFailed storePath errMsg)-  for_ finishedJobs pushFinished+  updatedJobs <- applyPushJobUpdates pushIds (\_ -> markStorePath storePath)+  sendStorePathEvent (map PushJob.pushId updatedJobs) msg+  for_ (filter PushJob.isProcessed updatedJobs) pushFinished   removeStorePath storePath  -- Helpers++-- | Run a job operation without interleaving its state changes and events with+-- another job operation. STM keeps the internal state consistent; this lock+-- also keeps the externally visible event stream consistent with that state.+withJobLock :: PushManager a -> PushManager a+withJobLock action = do+  env@PushManagerEnv {pmJobLock} <- ask+  liftIO $ MVar.withMVar pmJobLock $ \_ -> runPushManager env action++sendStorePathEventForActiveJobsAt :: (Foldable f) => UTCTime -> f Protocol.PushRequestId -> PushEventMessage -> PushManager ()+sendStorePathEventForActiveJobsAt timestamp pushIds msg = withJobLock $ do+  pushJobs <- asks pmPushJobs >>= liftIO . readTVarIO+  let isActive pushId = maybe False (not . PushJob.isProcessed) $ HashMap.lookup pushId pushJobs+  sendStorePathEventAt timestamp (filter isActive $ toList pushIds) msg  storeToFilePath :: (MonadIO m) => Store -> StorePath -> m FilePath storeToFilePath store storePath = do
src/Cachix/Daemon/Types/PushManager.hs view
@@ -78,6 +78,9 @@     pmProgressEmitIntervalNs :: Word64,     -- | The number of pending (uncompleted) jobs.     pmPendingJobCount :: TVar Int,+    -- | Serializes job transitions with their externally visible events and+    -- store path index updates.+    pmJobLock :: MVar (),     -- | Manager for batching narinfo queries     pmNarinfoQueryManager :: NarinfoQueryManager Protocol.PushRequestId,     -- | Latch to coordinate graceful shutdown of the push pipeline
src/Cachix/Deploy/Activate.hs view
@@ -208,7 +208,7 @@           officialCache = "cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY="           substituters = ["--option", "extra-substituters", URI.serialize cacheURI]           noNegativeCaching = ["--option", "narinfo-cache-negative-ttl", "0"]-          sigs = ["--option", "trusted-public-keys", officialCache <> " " <> toS hostname <> "-1:" <> toS (WSS.publicKey cache)]+          sigs = ["--option", "extra-trusted-public-keys", officialCache <> " " <> toS hostname <> "-1:" <> toS (WSS.publicKey cache)]        in substituters ++ sigs ++ noNegativeCaching  runShell :: Log.LogStream -> FilePath -> [String] -> IO ()
src/Cachix/Deploy/Websocket.hs view
@@ -66,6 +66,11 @@   }   deriving (Show) +data WebSocketConnectionTimeout = WebSocketConnectionTimeout+  deriving (Eq, Show)++instance Exception WebSocketConnectionTimeout+ -- | A more ergonomic version of the Websocket 'Message' data type data Message msg   = ControlMessage WS.ControlMessage@@ -174,10 +179,25 @@ runClientWith :: Options -> WS.Connection.ConnectionOptions -> WS.ClientApp a -> IO a runClientWith Options {host, port, path, headers, useSSL} connectionOptions app =   if useSSL-    then Wuss.runSecureClientWith hostS (fromIntegral (URI.portNumber port)) (toS path) connectionOptions headers app+    then+      withConnectionTimeout+        timeoutMicroseconds+        (Wuss.newSecureClientConnectionWith hostS (fromIntegral (URI.portNumber port)) (toS path) connectionOptions headers)+        app     else WS.runClientWith hostS (URI.portNumber port) (toS path) connectionOptions headers app   where     hostS = toS (URI.hostBS host)+    timeoutMicroseconds = WS.Connection.connectionTimeout connectionOptions * 1000 * 1000++-- | Bound connection establishment without applying the timeout to the+-- long-running client application. The acquisition action returns its cleanup+-- action so that a successfully established connection is always closed.+withConnectionTimeout :: Int -> IO (a, IO ()) -> (a -> IO b) -> IO b+withConnectionTimeout timeoutMicroseconds acquire app = do+  connection <- Timeout.timeout timeoutMicroseconds acquire+  case connection of+    Nothing -> throwIO WebSocketConnectionTimeout+    Just (resource, close) -> Safe.bracket (pure resource) (const close) app  -- Handle JSON messages
+ test/CommandConfigSpec.hs view
@@ -0,0 +1,14 @@+module CommandConfigSpec (spec) where++import Cachix.Client.Command.Config (generateKeypairAccessDenied)+import Protolude+import Test.Hspec++spec :: Spec+spec =+  describe "generateKeypairAccessDenied" $ do+    it "explains that signing-key creation requires a personal administrator token" $ do+      let message = displayException $ generateKeypairAccessDenied "corestory"+      message `shouldContain` "Cannot create a signing key for binary cache corestory."+      message `shouldContain` "requires a personal auth token"+      message `shouldContain` "Per-cache read/write tokens"
test/Daemon/PushManagerSpec.hs view
@@ -9,10 +9,12 @@ import Cachix.Daemon.Push qualified as Daemon.Push import Cachix.Daemon.PushManager import Cachix.Daemon.PushManager.PushJob qualified as PushJob+import Cachix.Daemon.Types.PushEvent (PushEvent (..), PushEventMessage (..)) import Cachix.Daemon.Types.PushManager import Cachix.Types.BinaryCache qualified as BinaryCache import Cachix.Types.Permission (Permission (Write))-import Control.Concurrent.Async (concurrently_)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar qualified as MVar import Control.Concurrent.STM.TVar import Control.Monad (fail) import Control.Retry (defaultRetryStatus)@@ -22,6 +24,7 @@ import Protolude import Servant.Auth.Client (Token (Token)) import System.IO.Temp (withSystemTempDirectory)+import System.Timeout qualified as Timeout import Test.Hspec  instance MonadFail PushManager where@@ -147,6 +150,223 @@               prSkippedPaths = mempty             } +    describe "failing jobs" $ do+      it "notifies subscribers when a job fails before closure resolution" $ do+        events <- newTVarIO []+        withPushManagerOnEvent (recordEvents events) $ \pm -> do+          runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"], Protocol.subscribeToUpdates = True}+            Just pushId <- addPushJobFromRequest request+            failPushJob pushId "closure resolution failed"++            withPushJob pushId $ \pushJob ->+              liftIO $ PushJob.status pushJob `shouldBe` Failed++          messages <- map eventMessage . reverse <$> readTVarIO events+          messages+            `shouldBe` [ PushStorePathFailed "bar" "closure resolution failed",+                         PushStorePathFailed "foo" "closure resolution failed",+                         PushFinished+                       ]++      it "does not emit further events for failed jobs on shutdown" $ do+        events <- newTVarIO []+        withPushManagerOnEvent (recordEvents events) $ \pm -> do+          runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = True}+            Just pushId <- addPushJobFromRequest request+            failPushJob pushId "closure resolution failed"++          eventsAfterFailure <- readTVarIO events+          failed <- runPushManager pm $ failPendingJobs "daemon stopped"+          length failed `shouldBe` 0+          eventsAfterShutdown <- readTVarIO events+          eventsAfterShutdown `shouldBe` eventsAfterFailure++      it "only reports paths that were not pushed when failing a resolved job" $ do+        events <- newTVarIO []+        withPushManagerOnEvent (recordEvents events) $ \pm -> do+          runPushManager pm $ do+            let paths = ["bar", "foo"]+                pathSet = Set.fromList paths+            let request = Protocol.PushRequest {Protocol.storePaths = paths, Protocol.subscribeToUpdates = True}+            Just pushId <- addPushJobFromRequest request+            resolvePushJob pushId (PushJob.ResolvedClosure pathSet pathSet)+            pushStorePathDone "bar"+            void $ failPendingJobs "daemon stopped"++          messages <- map eventMessage . reverse <$> readTVarIO events+          messages+            `shouldBe` [ PushStarted,+                         PushStorePathDone "bar",+                         PushStorePathFailed "foo" "daemon stopped",+                         PushFinished+                       ]++      it "does not emit events for a failed job when its in-flight paths finish" $ do+        events <- newTVarIO []+        withPushManagerOnEvent (recordEvents events) $ \pm -> do+          runPushManager pm $ do+            let paths = ["bar", "foo"]+                pathSet = Set.fromList paths+            let request = Protocol.PushRequest {Protocol.storePaths = paths, Protocol.subscribeToUpdates = True}+            Just pushId <- addPushJobFromRequest request+            resolvePushJob pushId (PushJob.ResolvedClosure pathSet pathSet)+            failPushJob pushId "daemon stopped"++          eventsAfterFailure <- readTVarIO events+          lastMay (map eventMessage (reverse eventsAfterFailure)) `shouldBe` Just PushFinished++          -- The uploads that were in flight when the job failed report back.+          runPushManager pm $ do+            pushStorePathAttempt "foo" 1 defaultRetryStatus+            pushStorePathProgress "foo" 1 1+            pushStorePathDone "foo"+            pushStorePathFailed "bar" "upload failed"++          eventsAfterUploads <- readTVarIO events+          eventsAfterUploads `shouldBe` eventsAfterFailure++      it "ignores closure resolution for a job that already failed" $ do+        events <- newTVarIO []+        withPushManagerOnEvent (recordEvents events) $ \pm -> do+          pushId <- runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = True}+            Just pushId <- addPushJobFromRequest request+            failPushJob pushId "daemon stopped"+            pure pushId++          eventsAfterFailure <- readTVarIO events+          lastMay (map eventMessage (reverse eventsAfterFailure)) `shouldBe` Just PushFinished++          runPushManager pm $ do+            let pathSet = Set.fromList ["foo"]+            resolvePushJob pushId (PushJob.ResolvedClosure pathSet pathSet)+            pushStorePathDone "foo"++            withPushJob pushId $ \pushJob ->+              liftIO $ PushJob.status pushJob `shouldBe` Failed+            count <- pendingJobCount+            liftIO $ count `shouldBe` 0++          eventsAfterResolve <- readTVarIO events+          eventsAfterResolve `shouldBe` eventsAfterFailure++      it "serializes closure resolution with failure" $ do+        events <- newTVarIO []+        startedEvent <- MVar.newEmptyMVar+        releaseStartedEvent <- MVar.newEmptyMVar+        let onEvent pushId event = do+              recordEvents events pushId event+              when (eventMessage event == PushStarted) $ do+                MVar.putMVar startedEvent ()+                MVar.takeMVar releaseStartedEvent++        withPushManagerOnEvent onEvent $ \pm -> do+          pushId <- runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = True}+            Just pushId <- addPushJobFromRequest request+            pure pushId++          let pathSet = Set.singleton "foo"+              resolve = runPushManager pm $ resolvePushJob pushId (PushJob.ResolvedClosure pathSet pathSet)+              failJob = runPushManager pm $ failPushJob pushId "daemon stopped"+          Async.withAsync resolve $ \resolveThread -> do+            MVar.takeMVar startedEvent+            failureInvoked <- MVar.newEmptyMVar+            Async.withAsync (MVar.putMVar failureInvoked () >> failJob) $ \failureThread -> do+              MVar.takeMVar failureInvoked+              assertStillRunning failureThread+              MVar.putMVar releaseStartedEvent ()+              Async.wait resolveThread+              Async.wait failureThread++          runPushManager pm $ do+            pushStorePathAttempt "foo" 1 defaultRetryStatus+            pushStorePathProgress "foo" 1 1+            pushStorePathDone "foo"++          messages <- map eventMessage . reverse <$> readTVarIO events+          messages+            `shouldBe` [ PushStarted,+                         PushStorePathFailed "foo" "daemon stopped",+                         PushFinished+                       ]++      it "serializes in-flight progress events with failure" $ do+        events <- newTVarIO []+        progressEvent <- MVar.newEmptyMVar+        releaseProgressEvent <- MVar.newEmptyMVar+        let onEvent pushId event = do+              recordEvents events pushId event+              when (isProgressEvent event) $ do+                MVar.putMVar progressEvent ()+                MVar.takeMVar releaseProgressEvent++        withPushManagerOnEvent onEvent $ \pm -> do+          pushId <- runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = True}+                pathSet = Set.singleton "foo"+            Just pushId <- addPushJobFromRequest request+            resolvePushJob pushId (PushJob.ResolvedClosure pathSet pathSet)+            pure pushId++          let failJob = runPushManager pm $ failPushJob pushId "daemon stopped"+          Async.withAsync (runPushManager pm $ pushStorePathProgress "foo" 1 1) $ \progressThread -> do+            MVar.takeMVar progressEvent+            failureInvoked <- MVar.newEmptyMVar+            Async.withAsync (MVar.putMVar failureInvoked () >> failJob) $ \failureThread -> do+              MVar.takeMVar failureInvoked+              assertStillRunning failureThread+              MVar.putMVar releaseProgressEvent ()+              Async.wait progressThread+              Async.wait failureThread++          runPushManager pm $ do+            pushStorePathAttempt "foo" 1 defaultRetryStatus+            pushStorePathDone "foo"++          messages <- map eventMessage . reverse <$> readTVarIO events+          messages+            `shouldBe` [ PushStarted,+                         PushStorePathProgress "foo" 1 1,+                         PushStorePathFailed "foo" "daemon stopped",+                         PushFinished+                       ]++      it "keeps shared paths active for jobs that have not failed" $ do+        events <- newTVarIO []+        withPushManagerOnEvent (recordEvents events) $ \pm -> do+          (failedId, completedId) <- runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = True}+                pathSet = Set.singleton "foo"+                closure = PushJob.ResolvedClosure pathSet pathSet+            Just failedId <- addPushJobFromRequest request+            Just completedId <- addPushJobFromRequest request+            resolvePushJob failedId closure+            resolvePushJob completedId closure+            failPushJob failedId "daemon stopped"+            pushStorePathDone "foo"+            pure (failedId, completedId)++          recordedEvents <- reverse <$> readTVarIO events+          messagesFor failedId recordedEvents+            `shouldBe` [ PushStarted,+                         PushStorePathFailed "foo" "daemon stopped",+                         PushFinished+                       ]+          messagesFor completedId recordedEvents+            `shouldBe` [ PushStarted,+                         PushStorePathDone "foo",+                         PushFinished+                       ]++          failedJob <- runPushManager pm $ lookupPushJob failedId+          PushJob.status <$> failedJob `shouldBe` Just Failed+          completedJob <- runPushManager pm $ lookupPushJob completedId+          completedJob `shouldBe` Nothing+          runPushManager pm pendingJobCount `shouldReturn` 0+     describe "graceful shutdown" $ do       it "shuts down with no jobs" $         withPushManager $ \pm -> do@@ -166,7 +386,7 @@           return pushId          startTime <- getCurrentTime-        concurrently_ (drainPushManager longTimeoutOptions pm) $+        Async.concurrently_ (drainPushManager longTimeoutOptions pm) $           runPushManager pm $             for_ paths pushStorePathDone         endTime <- getCurrentTime@@ -192,7 +412,10 @@         result `shouldBe` False  withPushManager :: (PushManagerEnv -> IO a) -> IO a-withPushManager f = do+withPushManager = withPushManagerOnEvent mempty++withPushManagerOnEvent :: OnPushEvent -> (PushManagerEnv -> IO a) -> IO a+withPushManagerOnEvent onPushEvent f = do   CNix.init   withTempStore $ \store -> do     logger <- liftIO $ Log.new "daemon" Nothing Log.Debug@@ -203,7 +426,22 @@         pushOptions = defaultPushOptions         batchOptions = defaultNarinfoQueryOptions         pushParams = Daemon.Push.newPushParams store clientEnv binaryCache pushSecret pushOptions-    newPushManagerEnv pushOptions batchOptions pushParams mempty logger >>= f+    newPushManagerEnv pushOptions batchOptions pushParams onPushEvent logger >>= f++recordEvents :: TVar [PushEvent] -> OnPushEvent+recordEvents events _ event = atomically $ modifyTVar' events (event :)++assertStillRunning :: Async.Async () -> IO ()+assertStillRunning thread = do+  result <- Timeout.timeout 100000 $ Async.wait thread+  result `shouldBe` Nothing++isProgressEvent :: PushEvent -> Bool+isProgressEvent PushEvent {eventMessage = PushStorePathProgress {}} = True+isProgressEvent _ = False++messagesFor :: Protocol.PushRequestId -> [PushEvent] -> [PushEventMessage]+messagesFor pushId = map eventMessage . filter ((== pushId) . eventPushId)  inPushManager :: PushManager a -> IO a inPushManager f = withPushManager (`runPushManager` f)
test/DeploySpec.hs view
@@ -5,6 +5,8 @@ import Cachix.Deploy.Lock (withTryLock, withTryLockAndPid) import Cachix.Deploy.Log qualified as Log import Cachix.Deploy.OptionsParser qualified as CLI+import Cachix.Deploy.Websocket qualified as WebSocket+import Control.Concurrent.MVar qualified as MVar import Control.Retry qualified as Retry import Protolude import System.IO.Temp (withSystemTempDirectory)@@ -33,6 +35,19 @@             void $ withTryLockAndPid (lockFile agent) (pidFile agent) $ do               mpid <- waitForAgent retryPolicy agent               mpid `shouldSatisfy` isJust++    describe "WebSocket connection timeout" $ do+      it "interrupts stalled connection establishment" $ do+        let stalledConnection = do+              void (MVar.newEmptyMVar >>= MVar.takeMVar :: IO ())+              pure ((), pure ())+        WebSocket.withConnectionTimeout (10 * 1000) stalledConnection pure+          `shouldThrow` (== WebSocket.WebSocketConnectionTimeout)++      it "closes an established connection after the client exits" $ do+        closed <- MVar.newEmptyMVar+        WebSocket.withConnectionTimeout (10 * 1000) (pure ((), MVar.putMVar closed ())) pure+        MVar.tryTakeMVar closed `shouldReturn` Just ()  withTestAgent :: FilePath -> (Agent -> IO ()) -> IO () withTestAgent tempDir action = do