diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/cachix.cabal b/cachix.cabal --- a/cachix.cabal +++ b/cachix.cabal @@ -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 diff --git a/src/Cachix/Client/Command/Config.hs b/src/Cachix/Client/Command/Config.hs --- a/src/Cachix/Client/Command/Config.hs +++ b/src/Cachix/Client/Command/Config.hs @@ -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. + |] diff --git a/src/Cachix/Daemon/PushManager.hs b/src/Cachix/Daemon/PushManager.hs --- a/src/Cachix/Daemon/PushManager.hs +++ b/src/Cachix/Daemon/PushManager.hs @@ -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 diff --git a/src/Cachix/Daemon/Types/PushManager.hs b/src/Cachix/Daemon/Types/PushManager.hs --- a/src/Cachix/Daemon/Types/PushManager.hs +++ b/src/Cachix/Daemon/Types/PushManager.hs @@ -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 diff --git a/src/Cachix/Deploy/Activate.hs b/src/Cachix/Deploy/Activate.hs --- a/src/Cachix/Deploy/Activate.hs +++ b/src/Cachix/Deploy/Activate.hs @@ -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 () diff --git a/src/Cachix/Deploy/Websocket.hs b/src/Cachix/Deploy/Websocket.hs --- a/src/Cachix/Deploy/Websocket.hs +++ b/src/Cachix/Deploy/Websocket.hs @@ -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 diff --git a/test/CommandConfigSpec.hs b/test/CommandConfigSpec.hs new file mode 100644 --- /dev/null +++ b/test/CommandConfigSpec.hs @@ -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" diff --git a/test/Daemon/PushManagerSpec.hs b/test/Daemon/PushManagerSpec.hs --- a/test/Daemon/PushManagerSpec.hs +++ b/test/Daemon/PushManagerSpec.hs @@ -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) diff --git a/test/DeploySpec.hs b/test/DeploySpec.hs --- a/test/DeploySpec.hs +++ b/test/DeploySpec.hs @@ -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