cachix 1.2 → 1.3
raw patch · 14 files changed
+432/−154 lines, 14 filesdep +conduit-concurrent-mapdep +deepseqdep +network-uri
Dependencies added: conduit-concurrent-map, deepseq, network-uri
Files
- CHANGELOG.md +19/−2
- cachix-deployment/Main.hs +3/−1
- cachix.cabal +7/−1
- src/Cachix/Client/Commands.hs +44/−28
- src/Cachix/Client/Env.hs +12/−1
- src/Cachix/Client/InstallationMode.hs +1/−1
- src/Cachix/Client/OptionsParser.hs +1/−1
- src/Cachix/Client/Push.hs +71/−71
- src/Cachix/Client/Push/S3.hs +116/−0
- src/Cachix/Client/PushQueue.hs +19/−17
- src/Cachix/Deploy/ActivateCommand.hs +30/−7
- src/Cachix/Deploy/Agent.hs +37/−23
- src/Cachix/Deploy/Websocket.hs +1/−1
- src/Data/Conduit/ByteString.hs +71/−0
CHANGELOG.md view
@@ -5,6 +5,23 @@ 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] - 2023-03-06++### Added++- Upload nars using multiple parts, improving bandwidth speed and parallelization.++### Changed++- Bump defaults jobs to 8.+- Improve 401 erros by showing the body of backend response.+- Deploy: improve agent startup and shutdown.++### Fixed++- Unblock interrupt signal on darwin (possibly also Linux).+- Deploy: wait for the logs to finish when activating.+ ## [1.2] - 2023-01-06 ### Added@@ -15,11 +32,11 @@ - `cachix deploy activate` now by default waits for the agents to be deployed, displays the logs and exists if any deployments fail. If you'd like to keep the old behaviour pass `--async` flag. -## Changed+### Changed - We no longer pin Nix to speed up version bumps of Nix -## Fixed+### Fixed - A number of improvements to stability of the websocket connection used in cachix deploy.
cachix-deployment/Main.hs view
@@ -149,8 +149,9 @@ Activate.Failure e -> logDeploymentFailed e Activate.Rollback e -> logDeploymentFailed e Activate.Success -> do+ -- NOTE: the activate command uses this message to detect the end of the log Log.streamLine logStream "Successfully activated the deployment."- withLog $ K.logLocM K.InfoS $ K.ls $ "Deployment #" <> deploymentIndex <> " finished"+ withLog $ K.logLocM K.InfoS $ K.ls $ "Deployment #" <> deploymentIndex <> " finished." endDeployment activationStatus where@@ -176,6 +177,7 @@ Log.streamLine logStream $ toS $ unwords+ -- NOTE: the activate command uses this message to detect the end of the log [ "Failed to activate the deployment.", toS $ displayException e ]
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix-version: 1.2+version: 1.3 license: Apache-2.0 license-file: LICENSE copyright: 2018 Domen Kozar@@ -22,6 +22,7 @@ common defaults default-extensions: NoImplicitPrelude+ NoImportQualifiedPost DeriveAnyClass DeriveGeneric DerivingVia@@ -60,6 +61,7 @@ Cachix.Client.NixVersion Cachix.Client.OptionsParser Cachix.Client.Push+ Cachix.Client.Push.S3 Cachix.Client.PushQueue Cachix.Client.Retry Cachix.Client.Secrets@@ -77,6 +79,7 @@ Cachix.Deploy.StdinProcess Cachix.Deploy.Websocket Cachix.Deploy.WebsocketPong+ Data.Conduit.ByteString System.Nix.Base32 hs-source-dirs: src@@ -91,11 +94,13 @@ , cachix-api , concurrent-extra , conduit >=1.3.0+ , conduit-concurrent-map , conduit-extra , conduit-zstd , containers , cookie , cryptonite+ , deepseq , dhall >=1.28.0 , directory , ed25519@@ -118,6 +123,7 @@ , memory , mmorph , netrc+ , network-uri , optparse-applicative , pretty-terminal , prettyprinter
src/Cachix/Client/Commands.hs view
@@ -37,6 +37,7 @@ import qualified Cachix.Client.WatchStore as WatchStore import qualified Cachix.Types.BinaryCache as BinaryCache import qualified Cachix.Types.SigningKeyCreate as SigningKeyCreate+import qualified Control.Concurrent.Async as Async import Control.Exception.Safe (throwM) import Control.Retry (RetryStatus (rsIterNumber)) import Crypto.Sign.Ed25519 (PublicKey (PublicKey), createKeypair)@@ -55,7 +56,6 @@ 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!@@ -106,11 +106,14 @@ notAuthenticatedBinaryCache :: Text -> CachixException notAuthenticatedBinaryCache name = AccessDeniedBinaryCache $- "Binary cache " <> name <> " doesn't exist or it's private. " <> Config.noAuthTokenError+ "Binary cache " <> name <> " doesn't exist or it's private and you need a token: " <> Config.noAuthTokenError -accessDeniedBinaryCache :: Text -> CachixException-accessDeniedBinaryCache name =- AccessDeniedBinaryCache $ "Binary cache " <> name <> " doesn't exist or it's private and you don't have access it"+accessDeniedBinaryCache :: Text -> Maybe ByteString -> CachixException+accessDeniedBinaryCache name maybeBody =+ AccessDeniedBinaryCache $ "Binary cache " <> name <> " doesn't exist or you don't have access." <> context maybeBody+ where+ context Nothing = ""+ context (Just body) = " Error: " <> toS body use :: Env -> Text -> InstallationMode.UseOptions -> IO () use env name useOptions = do@@ -119,12 +122,7 @@ -- 1. get cache public key res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient token name case res of- Left err- -- TODO: is checking for the existence of the config file the right thing to do here?- | isErr err status401 && isJust optionalAuthToken -> throwM $ accessDeniedBinaryCache name- | isErr err status401 -> throwM $ notAuthenticatedBinaryCache name- | isErr err status404 -> throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist."- | otherwise -> throwM err+ Left err -> handleCacheResponse name optionalAuthToken err Right binaryCache -> do () <- escalateAs UnsupportedNixVersion =<< assertNixVersion user <- InstallationMode.getUser@@ -140,6 +138,17 @@ InstallationMode.addBinaryCache (config env) binaryCache useOptions $ InstallationMode.getInstallationMode nixEnv useOptions +handleCacheResponse :: Text -> Maybe Token -> ClientError -> IO a+handleCacheResponse name optionalAuthToken err+ | isErr err status401 && isJust optionalAuthToken = throwM $ accessDeniedBinaryCache name (failureResponseBody err)+ | isErr err status401 = throwM $ notAuthenticatedBinaryCache name+ | isErr err status404 = throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist."+ | otherwise = throwM err++failureResponseBody :: ClientError -> Maybe ByteString+failureResponseBody (FailureResponse _ response) = Just $ toS $ responseBody response+failureResponseBody _ = Nothing+ push :: Env -> PushArguments -> IO () push env (PushPaths opts name cliPaths) = do hasStdin <- not <$> hIsTerminalDevice stdin@@ -184,17 +193,32 @@ watchExec :: Env -> PushOptions -> Text -> Text -> [Text] -> IO () watchExec env pushOpts name cmd args = withPushParams env pushOpts name $ \pushParams -> do stdoutOriginal <- hDuplicate stdout- let process = (System.Process.proc (toS cmd) (toS <$> args)) {System.Process.std_out = System.Process.UseHandle stdoutOriginal}+ let process =+ (System.Process.proc (toS cmd) (toS <$> args))+ { System.Process.std_out = System.Process.UseHandle stdoutOriginal+ } watch = do hDuplicateTo stderr stdout -- redirect all stdout to stderr WatchStore.startWorkers (pushParamsStore pushParams) (numJobs pushOpts) pushParams- (_, exitCode) <- concurrently watch $ do- (_, _, _, processHandle) <- System.Process.createProcess process- exitCode <- System.Process.waitForProcess processHandle- Signals.raiseSignal Signals.sigINT- return exitCode- exitWith exitCode + 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+ where+ getProcessHandle (_, _, _, processHandle) = processHandle+ retryText :: RetryStatus -> Text retryText retrystatus = if rsIterNumber retrystatus == 0@@ -205,10 +229,7 @@ pushStrategy store authToken opts name compressionMethod storePath = PushStrategy { onAlreadyPresent = pass,- on401 =- if isJust authToken- then throwM (accessDeniedBinaryCache name)- else throwM (notAuthenticatedBinaryCache name),+ on401 = handleCacheResponse name authToken, onError = throwM, onAttempt = \retrystatus size -> do path <- decodeUtf8With lenientDecode <$> storePathToPath store storePath@@ -230,12 +251,7 @@ let token = fromMaybe (Token "") authToken res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient token name case res of- Left err- -- TODO: is checking for the existence of the config file the right thing to do here?- | isErr err status401 && isJust authToken -> throwM $ accessDeniedBinaryCache name- | isErr err status401 -> throwM $ notAuthenticatedBinaryCache name- | isErr err status404 -> throwM $ BinaryCacheNotFound $ "Binary cache " <> name <> " does not exist."- | otherwise -> throwM err+ Left err -> handleCacheResponse name authToken err Right binaryCache -> pure (Just $ BinaryCache.preferredCompressionMethod binaryCache) let compressionMethod = fromMaybe BinaryCache.ZSTD (head $ catMaybes [Cachix.Client.OptionsParser.compressionMethod pushOpts, compressionMethodBackend]) withStore $ \store ->
src/Cachix/Client/Env.hs view
@@ -12,7 +12,7 @@ import Cachix.Client.URI (getBaseUrl) import Cachix.Client.Version (cachixVersion) import qualified Hercules.CNix as CNix-import Hercules.CNix.Store (Store, openStore)+import qualified Hercules.CNix.Util as CNix.Util import Network.HTTP.Client ( ManagerSettings, managerModifyRequest,@@ -25,6 +25,7 @@ import Protolude.Conv import Servant.Client.Streaming (ClientEnv, mkClientEnv) import System.Directory (canonicalizePath)+import System.Posix.Signals (getSignalMask, setSignalMask) data Env = Env { cachixoptions :: Config.CachixOptions,@@ -34,7 +35,17 @@ mkEnv :: Options.Flags -> IO Env mkEnv flags = do+ signalset <- getSignalMask+ -- Initialize the Nix library CNix.init++ -- darwin: restore the signal mask modified by Nix+ -- https://github.com/cachix/cachix/issues/501+ setSignalMask signalset++ -- Interrupt Nix before throwing UserInterrupt+ CNix.Util.installDefaultSigINTHandler+ -- make sure path to the config is passed as absolute to dhall logic canonicalConfigPath <- canonicalizePath (Options.configPath flags) cfg <- Config.getConfig canonicalConfigPath
src/Cachix/Client/InstallationMode.hs view
@@ -90,7 +90,7 @@ b) Add the following to your configuration.nix to add your user as trusted and then try again: - nix.trustedUsers = [ "root" "${user}" ];+ nix.settings.trusted-users = [ "root" "${user}" ]; |] addBinaryCache _ _ _ UntrustedRequiresSudo = do
src/Cachix/Client/OptionsParser.hs view
@@ -152,7 +152,7 @@ <> short 'j' <> help "Number of threads used for pushing store paths." <> showDefault- <> value 4+ <> value 8 ) <*> switch (long "omit-deriver" <> help "Do not publish which derivations built the store paths.") push = (\opts cache f -> Push $ f opts cache) <$> pushOptions <*> nameArg <*> (pushPaths <|> pushWatchStore)
src/Cachix/Client/Push.hs view
@@ -30,11 +30,12 @@ import Cachix.API.Signing (fingerprint, passthroughHashSink, passthroughHashSinkB16, passthroughSizeSink) import qualified Cachix.Client.Config as Config import Cachix.Client.Exception (CachixException (..))+import qualified Cachix.Client.Push.S3 as Push.S3 import Cachix.Client.Retry (retryAll) import Cachix.Client.Secrets import Cachix.Client.Servant import qualified Cachix.Types.BinaryCache as BinaryCache-import qualified Cachix.Types.ByteStringStreaming+import qualified Cachix.Types.MultipartUpload as Multipart import qualified Cachix.Types.NarInfoCreate as Api import qualified Cachix.Types.NarInfoHash as NarInfoHash import Control.Concurrent.Async (mapConcurrently)@@ -44,7 +45,6 @@ import Control.Retry (RetryStatus) import Crypto.Sign.Ed25519 import qualified Data.ByteString.Base64 as B64-import Data.Coerce (coerce) import Data.Conduit import qualified Data.Conduit.Lzma as Lzma (compress) import qualified Data.Conduit.Zstd as Zstd (compress)@@ -86,7 +86,7 @@ { -- | Called when a path is already in the cache. onAlreadyPresent :: m r, onAttempt :: RetryStatus -> Int64 -> m (),- on401 :: m r,+ on401 :: ClientError -> m r, onError :: ClientError -> m r, onDone :: m r, compressionMethod :: BinaryCache.CompressionMethod,@@ -131,7 +131,7 @@ Right NoContent -> onAlreadyPresent strategy -- we're done as store path is already in the cache Left err | isErr err status404 -> uploadStorePath cache storePath retrystatus- | isErr err status401 -> on401 strategy+ | isErr err status401 -> on401 strategy err | otherwise -> onError strategy err getCacheAuthToken :: PushSecret -> Token@@ -139,7 +139,7 @@ getCacheAuthToken (PushSigningKey token _) = token uploadStorePath ::- (MonadMask m, MonadIO m) =>+ (MonadIO m) => -- | details for pushing to cache PushParams m r -> StorePath ->@@ -151,85 +151,85 @@ -- TODO: storePathText is redundant. Use storePath directly. storePathText <- liftIO $ Store.storePathToPath store storePath let (storeHash, storeSuffix) = splitStorePath $ toS storePathText- name = pushParamsName cache+ cacheName = pushParamsName cache+ authToken = getCacheAuthToken (pushParamsSecret cache) clientEnv = pushParamsClientEnv cache strategy = pushParamsStrategy cache storePath withCompressor = case compressionMethod strategy of BinaryCache.XZ -> defaultWithXzipCompressorWithLevel (compressionLevel strategy) BinaryCache.ZSTD -> defaultWithZstdCompressorWithLevel (compressionLevel strategy)+ cacheClientEnv =+ clientEnv+ { baseUrl = (baseUrl clientEnv) {baseUrlHost = toS cacheName <> "." <> baseUrlHost (baseUrl clientEnv)}+ }+ narSizeRef <- liftIO $ newIORef 0 fileSizeRef <- liftIO $ newIORef 0 narHashRef <- liftIO $ newIORef ("" :: ByteString) fileHashRef <- liftIO $ newIORef ("" :: ByteString)+ -- This should be a noop because storePathText came from a StorePath normalized <- liftIO $ Store.followLinksToStorePath store $ toS storePathText pathinfo <- liftIO $ Store.queryPathInfo store normalized- -- stream store path as xz compressed nar file- let storePathSize :: Int64- storePathSize = Store.validPathInfoNarSize pathinfo+ let storePathSize = Store.validPathInfoNarSize pathinfo onAttempt strategy retrystatus storePathSize- withCompressor $ \compressor -> do- let stream' =- streamNarIO narEffectsIO (toS storePathText) Data.Conduit.yield- .| passthroughSizeSink narSizeRef- .| passthroughHashSink narHashRef- .| compressor- .| passthroughSizeSink fileSizeRef- .| passthroughHashSinkB16 fileHashRef- let subdomain =- -- TODO: multipart- if (fromIntegral storePathSize / (1024 * 1024) :: Double) > 100- then "api"- else toS name- newClientEnv =- clientEnv- { baseUrl = (baseUrl clientEnv) {baseUrlHost = subdomain <> "." <> baseUrlHost (baseUrl clientEnv)}- }- (_ :: NoContent) <- liftIO $ do- (`withClientM` newClientEnv)- (API.createNar cachixClient (getCacheAuthToken (pushParamsSecret cache)) name (Just $ compressionMethod strategy) (mapOutput coerce stream'))- escalate- (_ :: NoContent) <- liftIO $ do- narSize <- readIORef narSizeRef- narHash <- ("sha256:" <>) . System.Nix.Base32.encode <$> readIORef narHashRef- narHashNix <- Store.validPathInfoNarHash32 pathinfo- when (narHash /= toS narHashNix) $ throwM $ NarHashMismatch $ toS storePathText <> ": Nar hash mismatch between nix-store --dump and nix db. You can repair db metadata by running as root: $ nix-store --verify --repair --check-contents"- fileHash <- readIORef fileHashRef- fileSize <- readIORef fileSizeRef- deriverPath <-- if omitDeriver strategy- then pure Nothing- else Store.validPathInfoDeriver store pathinfo- deriver <- for deriverPath Store.getStorePathBaseName- referencesPathSet <- Store.validPathInfoReferences store pathinfo- referencesPaths <- sort . fmap toS <$> for referencesPathSet (Store.storePathToPath store)- references <- sort . fmap toS <$> for referencesPathSet Store.getStorePathBaseName- let fp = fingerprint (decodeUtf8With lenientDecode storePathText) narHash narSize referencesPaths- (sig, authToken) = case pushParamsSecret cache of- PushToken token -> (Nothing, token)- PushSigningKey token signKey -> (Just $ toS $ B64.encode $ unSignature $ dsign (signingSecretKey signKey) fp, token)- nic =- Api.NarInfoCreate- { Api.cStoreHash = storeHash,- Api.cStoreSuffix = storeSuffix,- Api.cNarHash = narHash,- Api.cNarSize = narSize,- Api.cFileSize = fileSize,- Api.cFileHash = toS fileHash,- Api.cReferences = references,- Api.cDeriver = maybe "unknown-deriver" (decodeUtf8With lenientDecode) deriver,- Api.cSig = sig- }- escalate $ Api.isNarInfoCreateValid nic- -- Upload narinfo with signature- escalate <=< (`runClientM` clientEnv) $- API.createNarinfo- cachixClient- authToken- name- (NarInfoHash.NarInfoHash storeHash)- nic- onDone strategy++ withCompressor $ \compressor -> liftIO $ do+ uploadResult <-+ runConduitRes $+ streamNarIO narEffectsIO (toS storePathText) Data.Conduit.yield+ .| passthroughSizeSink narSizeRef+ .| passthroughHashSink narHashRef+ .| compressor+ .| passthroughSizeSink fileSizeRef+ .| passthroughHashSinkB16 fileHashRef+ .| Push.S3.streamUpload cacheClientEnv authToken cacheName (compressionMethod strategy)++ case uploadResult of+ Left err -> throwIO err+ Right (narId, uploadId, parts) -> do+ narSize <- readIORef narSizeRef+ narHash <- ("sha256:" <>) . System.Nix.Base32.encode <$> readIORef narHashRef+ narHashNix <- Store.validPathInfoNarHash32 pathinfo+ when (narHash /= toS narHashNix) $ throwM $ NarHashMismatch $ toS storePathText <> ": Nar hash mismatch between nix-store --dump and nix db. You can repair db metadata by running as root: $ nix-store --verify --repair --check-contents"+ fileHash <- readIORef fileHashRef+ fileSize <- readIORef fileSizeRef+ deriverPath <-+ if omitDeriver strategy+ then pure Nothing+ else Store.validPathInfoDeriver store pathinfo+ deriver <- for deriverPath Store.getStorePathBaseName+ referencesPathSet <- Store.validPathInfoReferences store pathinfo+ referencesPaths <- sort . fmap toS <$> for referencesPathSet (Store.storePathToPath store)+ references <- sort . fmap toS <$> for referencesPathSet Store.getStorePathBaseName+ let fp = fingerprint (decodeUtf8With lenientDecode storePathText) narHash narSize referencesPaths+ sig = case pushParamsSecret cache of+ PushToken _ -> Nothing+ PushSigningKey _ signKey -> Just $ toS $ B64.encode $ unSignature $ dsign (signingSecretKey signKey) fp+ nic =+ Api.NarInfoCreate+ { Api.cStoreHash = storeHash,+ Api.cStoreSuffix = storeSuffix,+ Api.cNarHash = narHash,+ Api.cNarSize = narSize,+ Api.cFileSize = fileSize,+ Api.cFileHash = toS fileHash,+ Api.cReferences = references,+ Api.cDeriver = maybe "unknown-deriver" (decodeUtf8With lenientDecode) deriver,+ Api.cSig = sig+ }+ escalate $ Api.isNarInfoCreateValid nic++ -- Complete the multipart upload and upload the narinfo+ let completeMultipartUploadRequest =+ API.completeNarUpload cachixClient authToken cacheName narId uploadId $+ Multipart.CompletedMultipartUpload+ { Multipart.parts = parts,+ Multipart.narInfoCreate = nic+ }+ void $ withClientM completeMultipartUploadRequest cacheClientEnv escalate++ onDone strategy -- | Push an entire closure --
+ src/Cachix/Client/Push/S3.hs view
@@ -0,0 +1,116 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE Rank2Types #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Cachix.Client.Push.S3 where++import qualified Cachix.API as API+import Cachix.API.Error+import Cachix.Client.Retry (retryAll)+import Cachix.Client.Servant (cachixClient)+import Cachix.Types.BinaryCache+import qualified Cachix.Types.MultipartUpload as Multipart+import Conduit (MonadResource, MonadUnliftIO)+import Control.DeepSeq (rwhnf)+import Crypto.Hash (Digest, MD5)+import qualified Crypto.Hash as Crypto+import Data.ByteArray.Encoding (Base (..), convertToBase)+import Data.Conduit (ConduitT, handleC, (.|))+import Data.Conduit.ByteString (ChunkSize, chunkStream)+import qualified Data.Conduit.Combinators as CC+import Data.Conduit.ConcurrentMap (concurrentMapM_)+import Data.List (lookup)+import qualified Data.List.NonEmpty as NonEmpty+import Data.UUID (UUID)+import Network.HTTP.Client as HTTP+import qualified Network.HTTP.Types.Header as HTTP+import Protolude+import Servant.Auth ()+import Servant.Auth.Client+import qualified Servant.Client as Client+import Servant.Client.Streaming+import Servant.Conduit ()++-- | The size of each uploaded part.+--+-- Common values for S3 are 8MB and 16MB. The minimum is 5MB.+--+-- Lower values will increase HTTP overhead. Some cloud services impose request body limits.+-- For example, Amazon API Gateway caps out at 10MB.+chunkSize :: ChunkSize+chunkSize = 8 * 1024 * 1024++-- | The number of parts to upload concurrently.+-- Speeds up the upload of very large files.+concurrentParts :: Int+concurrentParts = 8++-- | The size of the temporary output buffer.+--+-- Keep this value high to avoid stalling smaller uploads while waiting for a large upload to complete.+-- Each completed upload response is very lightweight.+outputBufferSize :: Int+outputBufferSize = 100++streamUpload ::+ forall m.+ (MonadUnliftIO m, MonadResource m) =>+ ClientEnv ->+ Token ->+ Text ->+ CompressionMethod ->+ ConduitT+ ByteString+ Void+ m+ (Either SomeException (UUID, Text, Maybe (NonEmpty Multipart.CompletedPart)))+streamUpload env authToken cacheName compressionMethod = do+ Multipart.CreateMultipartUploadResponse {narId, uploadId} <- createMultipartUpload++ handleC (abortMultipartUpload narId uploadId) $+ chunkStream (Just chunkSize)+ .| concurrentMapM_ concurrentParts outputBufferSize (uploadPart narId uploadId)+ .| completeMultipartUpload narId uploadId+ where+ manager = Client.manager env++ createMultipartUpload :: ConduitT ByteString Void m Multipart.CreateMultipartUploadResponse+ createMultipartUpload =+ liftIO $ withClientM createNarRequest env escalate+ where+ createNarRequest = API.createNar cachixClient authToken cacheName (Just compressionMethod)++ uploadPart :: UUID -> Text -> (Int, ByteString) -> m (Maybe Multipart.CompletedPart)+ uploadPart narId uploadId (partNumber, !part) = do+ let partHashMD5 :: Digest MD5 = Crypto.hash part+ contentMD5 :: ByteString = convertToBase Base64 partHashMD5++ let uploadNarPartRequest = API.uploadNarPart cachixClient authToken cacheName narId uploadId partNumber (Multipart.SigningData (decodeUtf8 contentMD5))+ Multipart.UploadPartResponse {uploadUrl} <- liftIO $ withClientM uploadNarPartRequest env escalate++ initialRequest <- liftIO $ HTTP.parseUrlThrow (toS uploadUrl)+ let request =+ initialRequest+ { HTTP.method = "PUT",+ HTTP.requestBody = HTTP.RequestBodyBS part,+ HTTP.requestHeaders =+ [ ("Content-Type", "application/octet-stream"),+ ("Content-MD5", contentMD5)+ ]+ }++ response <- liftIO $ retryAll $ \_ -> HTTP.httpNoBody request manager+ let eTag = decodeUtf8 <$> lookup HTTP.hETag (HTTP.responseHeaders response)+ -- Strictly evaluate each eTag after uploading each part+ let !_ = rwhnf eTag+ return $ Multipart.CompletedPart partNumber <$> eTag++ completeMultipartUpload narId uploadId = do+ parts <- CC.sinkList+ return $ Right (narId, uploadId, sequenceA $ NonEmpty.fromList parts)++ abortMultipartUpload narId uploadId err = do+ let abortMultipartUploadRequest = API.abortMultipartUpload cachixClient authToken cacheName narId uploadId+ _ <- liftIO $ withClientM abortMultipartUploadRequest env escalate+ return $ Left err
src/Cachix/Client/PushQueue.hs view
@@ -16,14 +16,12 @@ 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@@ -67,12 +65,10 @@ progress <- newTVarIO 0 let pushWorkerState = PushWorkerState newPushQueue progress pushWorker <- async $ replicateConcurrently_ numWorkers $ worker pushParams pushWorkerState- let signalset =- Signals.CatchOnce $- exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState- void $ Signals.installHandler Signals.sigINT signalset Nothing- void $ Signals.installHandler Signals.sigTERM signalset Nothing- (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker]+ let drainQueue =+ exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState++ (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker] `finally` drainQueue case eitherException of Left exc | fromException exc == Just StopWorker -> return () Left exc -> throwIO exc@@ -103,15 +99,22 @@ queryLoop (workerState {alreadyQueued = S.union missingStorePathsSet alreadyQueuedSet}) pushqueue pushParams -- | Stop watching the store and push all pending store paths.--- This should only be used once per process. exitOnceQueueIsEmpty :: IO () -> Async () -> Async () -> QueryWorkerState -> PushWorkerState -> IO ()-exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState =- join . once $ do- putTextError "Stopped watching /nix/store and waiting for queue to empty ..."- void Systemd.notifyStopping- stopProducerCallback- go+exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = 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 where+ stopWorkers = do+ cancelWith queryWorker StopWorker+ cancelWith pushWorker StopWorker+ go = do (isDone, inprogress, queueLength) <- atomically $ do pushQueueLength <- TBQueue.lengthTBQueue $ pushQueue pushWorkerState@@ -122,9 +125,8 @@ return (isDone, inprogress, pushQueueLength) if isDone then do+ stopWorkers putTextError "Done."- cancelWith queryWorker StopWorker- cancelWith pushWorker StopWorker else do -- extend shutdown for another 90s void $ Systemd.notify False $ "EXTEND_TIMEOUT_USEC=" <> show (90 * 1000 * 1000 :: Int)
src/Cachix/Deploy/ActivateCommand.hs view
@@ -31,6 +31,8 @@ import Servant.Client.Streaming (ClientEnv, runClientM) import Servant.Conduit () import System.Environment (getEnv)+import qualified Text.Megaparsec as Parse+import qualified Text.Megaparsec.Char as Parse run :: Env.Env -> DeployOptions.ActivateOptions -> IO () run env DeployOptions.ActivateOptions {DeployOptions.payloadPath, DeployOptions.agents, DeployOptions.deployAsync} = do@@ -60,6 +62,7 @@ Text.putStr (renderOverview agents) + -- Skip streaming the logs when run with the --async flag when deployAsync exitSuccess Text.putStr "\n\n"@@ -93,10 +96,16 @@ WebSocket.identifier = identifier } - deployment <-- Async.withAsync (printLogsToTerminal options agentName) $ \_ ->- pollDeploymentStatus clientenv (Token agentToken) deploymentID+ deployment <- Async.withAsync (printLogsToTerminal options agentName) $ \logThread -> do+ deployment <- pollDeploymentStatus clientenv (Token agentToken) deploymentID + -- Wait for all the logs to arrive+ let status = Deployment.status deployment+ when (status == Deployment.Failed || status == Deployment.Succeeded) $+ void (Async.waitCatch logThread)++ pure deployment+ pure (agentName, deployment) pollDeploymentStatus :: ClientEnv -> Token -> UUID -> IO Deployment.Deployment@@ -116,14 +125,28 @@ threadDelay (2 * 1000 * 1000) loop -printLogsToTerminal :: WebSocket.Options -> Text -> IO a+printLogsToTerminal :: WebSocket.Options -> Text -> IO () printLogsToTerminal options agentName = WebSocket.runClientWith options WS.defaultConnectionOptions $ \connection ->- forever $ do+ fix $ \loop -> do message <- WS.receiveData connection case Aeson.eitherDecodeStrict' message of- Left error -> Text.putStrLn $ "Error parsing the log message: " <> show error- Right msg -> Text.putStrLn $ unwords [inBrackets agentName, WSS.line msg]+ Left error -> do+ Text.putStrLn $ "Error parsing the log message: " <> show error+ loop+ Right msg -> do+ putStrLn $ unwords [inBrackets agentName, WSS.line msg]+ unless (isDeploymentDone (WSS.line msg)) loop+ where+ -- Parse each log line looking for the success/failure messages.+ -- TODO: figure out a way to avoid this. How can we tell when the log is done?+ isDeploymentDone :: Text -> Bool+ isDeploymentDone = isRight . Parse.parse logEndMessageParser ""++ logEndMessageParser :: Parse.Parsec Void Text Text+ logEndMessageParser =+ Parse.string "Successfully activated the deployment"+ <|> Parse.string "Failed to activate the deployment" renderOverview :: [(Text, DeployResponse.Details)] -> Text renderOverview agents =
src/Cachix/Deploy/Agent.hs view
@@ -34,6 +34,7 @@ import qualified System.Posix.Signals as Signals import qualified System.Posix.Types as Posix import qualified System.Posix.User as Posix.User+import qualified System.Timeout as Timeout type ServiceWebSocket = WebSocket.WebSocket (WSS.Message WSS.AgentCommand) (WSS.Message WSS.BackendCommand) @@ -56,7 +57,7 @@ run :: Config.CachixOptions -> CLI.AgentOptions -> IO () run cachixOptions agentOptions = Log.withLog logOptions $ \withLog ->- logAndExitWithFailure withLog $+ logExceptions withLog $ withAgentLock agentOptions $ do checkUserOwnsHome @@ -83,9 +84,7 @@ channel <- WebSocket.receive websocket shutdownWebsocket <- connectToService websocket - let signalSet = Signals.CatchOnce shutdownWebsocket- void $ Signals.installHandler Signals.sigINT signalSet Nothing- void $ Signals.installHandler Signals.sigTERM signalSet Nothing+ installSignalHandlers shutdownWebsocket let agent = Agent@@ -109,20 +108,6 @@ agentName = CLI.name agentOptions profileName = fromMaybe "system" (CLI.profile agentOptions) - logAndExitWithFailure withLog action = withException action logException- where- logException :: SomeException -> IO ()- logException someE =- case fromException someE of- Nothing -> pure ()- Just ExitSuccess -> pure ()- Just e ->- withLog . K.logLocM K.ErrorS . K.ls $- unlines- [ "The agent encountered an exception:",- toS (displayException e)- ]- verbosity = if Config.verbose cachixOptions then Log.Verbose@@ -135,17 +120,44 @@ environment = "production" } +logExceptions :: Log.WithLog -> IO a -> IO a+logExceptions withLog action = withException action $ \someE -> do+ case fromException someE of+ Just ExitSuccess -> exitSuccess+ Just e -> do+ withLog . K.logLocM K.ErrorS . K.ls $+ unlines+ [ "The agent encountered an exception:",+ toS (displayException e)+ ]+ exitFailure+ Nothing -> exitFailure+ lockFilename :: Text -> FilePath lockFilename agentName = "agent-" <> toS agentName -- | Acquire a lock for this agent. Skip this step if we're bootstrapping the agent. withAgentLock :: CLI.AgentOptions -> IO () -> IO () withAgentLock CLI.AgentOptions {bootstrap = True} action = action-withAgentLock CLI.AgentOptions {name} action = do- lock <- Lock.withTryLockAndPid (lockFilename name) action- when (isNothing lock) $- throwIO (AgentAlreadyRunning name)+withAgentLock CLI.AgentOptions {name} action = tryToAcquireLock 0+ where+ tryToAcquireLock :: Int -> IO ()+ tryToAcquireLock attempts = do+ lock <- Lock.withTryLockAndPid (lockFilename name) action+ when (isNothing lock) $+ if attempts >= 5+ then throwIO (AgentAlreadyRunning name)+ else do+ threadDelay (3 * 1000 * 1000)+ tryToAcquireLock (attempts + 1) +installSignalHandlers :: IO () -> IO ()+installSignalHandlers shutdown =+ for_ [Signals.sigINT, Signals.sigTERM] $ \signal ->+ Signals.installHandler signal handler Nothing+ where+ handler = Signals.CatchOnce shutdown+ registerAgent :: Agent -> WSS.AgentInformation -> IO () registerAgent Agent {agentState, withLog} agentInformation = do withLog $ K.logLocM K.InfoS "Agent registered."@@ -228,7 +240,9 @@ -- Block until the initial connection is established void $ MVar.readMVar (WebSocket.connection websocket) - once $ MVar.putMVar close () >> Async.wait thread+ once $ do+ void $ MVar.tryPutMVar close ()+ void $ Timeout.timeout (5 * 1000 * 1000) (Async.wait thread) -- | Fetch the home directory and verify that the owner matches the current user. -- Throws either 'NoHomeFound' or 'UserDoesNotOwnHome'.
src/Cachix/Deploy/Websocket.hs view
@@ -316,7 +316,7 @@ threadDelay (seconds * 1000 * 1000) startGracePeriod :: IO a -> IO (Maybe a)-startGracePeriod = Timeout.timeout (5 * 1000 * 1000)+startGracePeriod = Timeout.timeout (3 * 1000 * 1000) -- | Try to gracefully close the WebSocket. --
+ src/Data/Conduit/ByteString.hs view
@@ -0,0 +1,71 @@+module Data.Conduit.ByteString where++import Conduit (MonadUnliftIO)+import qualified Data.ByteString as BS+import Data.ByteString.Internal (ByteString (PS), mallocByteString)+import Data.ByteString.Unsafe (unsafeIndex)+import Data.Conduit (ConduitT, await, yield, (.|))+import Foreign.ForeignPtr (ForeignPtr)+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)+import Foreign.Storable (pokeByteOff)+import Protolude hiding (hash, yield)++type ChunkSize = Int++minimumChunkSize :: ChunkSize+minimumChunkSize = 5 * 1024 * 1024++chunkStream :: (MonadUnliftIO m) => Maybe ChunkSize -> ConduitT ByteString (Int, ByteString) m ()+chunkStream mChunkSize = do+ processAndChunkOutputRaw chunkSize+ .| enumerateConduit+ where+ chunkSize :: ChunkSize+ chunkSize = maybe minimumChunkSize (max minimumChunkSize) mChunkSize++ -- count from 1+ enumerateConduit :: (Monad m) => ConduitT a (Int, a) m ()+ enumerateConduit = loop 1+ where+ loop i = await >>= maybe (return ()) (go i)+ go i x = do+ yield (i, x)+ loop (i + 1)+ {-# INLINE enumerateConduit #-}++data S = S (ForeignPtr Word8) (Ptr Word8) {-# UNPACK #-} !Int++newS :: ChunkSize -> IO S+newS chunkSize = do+ fptr <- mallocByteString chunkSize+ return (S fptr (unsafeForeignPtrToPtr fptr) 0)++processChunk :: ChunkSize -> ByteString -> S -> IO ([ByteString], S)+processChunk chunkSize input =+ loop identity 0+ where+ loop front idxIn s@(S fptr ptr idxOut)+ | idxIn >= BS.length input = return (front [], s)+ | otherwise = do+ pokeByteOff ptr idxOut (unsafeIndex input idxIn)+ let idxOut' = idxOut + 1+ idxIn' = idxIn + 1+ if idxOut' >= chunkSize+ then do+ let bs = PS fptr 0 idxOut'+ s' <- newS chunkSize+ loop (front . (bs :)) idxIn' s'+ else loop front idxIn' (S fptr ptr idxOut')++processAndChunkOutputRaw :: MonadIO m => ChunkSize -> ConduitT ByteString ByteString m ()+processAndChunkOutputRaw chunkSize =+ liftIO (newS chunkSize) >>= loop+ where+ loop s@(S fptr _ len) = do+ mbs <- await+ case mbs of+ Nothing -> yield $ PS fptr 0 len+ Just bs -> do+ (bss, s') <- liftIO $ processChunk chunkSize bs s+ mapM_ yield bss+ loop s'