cachix 1.7.1 → 1.7.2
raw patch · 11 files changed
+286/−147 lines, 11 files
Files
- CHANGELOG.md +19/−7
- cachix.cabal +1/−1
- src/Cachix/Client/Commands.hs +19/−16
- src/Cachix/Client/Commands/Push.hs +8/−6
- src/Cachix/Client/Daemon.hs +44/−43
- src/Cachix/Client/Daemon/Log.hs +4/−2
- src/Cachix/Client/Daemon/PushManager.hs +52/−24
- src/Cachix/Client/Daemon/Types/PushManager.hs +9/−4
- src/Cachix/Client/OptionsParser.hs +73/−7
- src/Cachix/Client/Push.hs +30/−7
- src/Cachix/Client/Push/S3.hs +27/−30
CHANGELOG.md view
@@ -5,9 +5,21 @@ 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.7.2] - 2024-03-06++### Added++- `--chunk-size`: defaults to 32MiB+- `--num-concurrent-chunks`: defaults to 4++### Fixed++- cachix push: allow pushing of big store paths (up to 320GB) by increasing chunk size to 32MB+- daemon: fix a bug where nothing would be pushed at all+ ## [1.7.1] - 2023-02-20 -## Fixed+### Fixed - daemon: add explicit sigINT/sigTERM handler - daemon: improve shutdown when jobs fail@@ -15,13 +27,13 @@ ## [1.7] - 2023-01-08 -## Added+### Added - daemon mode: push to cachix while building - `cachix import`: allow importing S3 binary caches into Cachix -## Fixed+### Fixed - Ignore sigPIPE exception @@ -37,7 +49,7 @@ ## [1.6.1] - 2023-09-25 -## Fixed+### Fixed - deploy: Correctly format `cachix deploy agent` exception messages @@ -53,17 +65,17 @@ - Set the file system encoding to utf8 to fix some hash mismatch errors -## Added+### Added - Implement a daemon for pushing store paths (more on this feature later on blog.cachix.org) ## [1.6] - 2023-06-27 -## Added+### Added - `cachix remove MYCACHE`: reverses `cachix use MYCACHE`. -## Changed+### Changed - `cachix push` now displays a progress bar and summary before pushing.
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix-version: 1.7.1+version: 1.7.2 synopsis: Command-line client for Nix binary cache hosting https://cachix.org
src/Cachix/Client/Commands.hs view
@@ -381,23 +381,28 @@ let daemonOptions = DaemonOptions {daemonSocketPath = Just daemonSock} daemon <- Daemon.new env daemonOptions (Just logHandle) pushOpts cacheName - -- Launch the daemon in the background- daemonThread <- Async.async $ Daemon.run daemon+ exitCode <-+ bracket (startDaemonThread daemon) (shutdownDaemonThread daemon logHandle) $ \_ -> do+ processEnv <- getEnvironment+ let newProcessEnv = Daemon.PostBuildHook.modifyEnv nixConfEnv processEnv+ let process =+ (System.Process.proc (toS cmd) (toS <$> args))+ { System.Process.std_out = System.Process.Inherit,+ System.Process.env = Just newProcessEnv,+ System.Process.delegate_ctlc = True+ }+ System.Process.withCreateProcess process $ \_ _ _ processHandle ->+ System.Process.waitForProcess processHandle - -- Subscribe to all push events+ exitWith exitCode+ where+ -- Launch the daemon in the background and subscribe to all push events+ startDaemonThread daemon = do+ daemonThread <- Async.async $ Daemon.run daemon daemonChan <- Daemon.subscribe daemon-- processEnv <- getEnvironment- let newProcessEnv = Daemon.PostBuildHook.modifyEnv nixConfEnv processEnv- let process =- (System.Process.proc (toS cmd) (toS <$> args))- { System.Process.std_out = System.Process.Inherit,- System.Process.env = Just newProcessEnv,- System.Process.delegate_ctlc = True- }- exitCode <- System.Process.withCreateProcess process $ \_ _ _ processHandle ->- System.Process.waitForProcess processHandle+ return (daemonThread, daemonChan) + shutdownDaemonThread daemon logHandle (daemonThread, daemonChan) = do -- TODO: process and fold events into a state during command execution daemonExitCode <- Async.withAsync (postWatchExec daemonChan) $ \_ -> do Daemon.stopIO daemon@@ -408,8 +413,6 @@ ExitFailure _ -> printLog logHandle ExitSuccess -> return () - exitWith exitCode- where postWatchExec chan = do statsRef <- newIORef HashMap.empty runConduit $
src/Cachix/Client/Commands/Push.hs view
@@ -15,10 +15,10 @@ import Cachix.Client.Env (Env (..)) import Cachix.Client.Exception (CachixException (..)) import Cachix.Client.HumanSize (humanSize)-import Cachix.Client.OptionsParser+import Cachix.Client.OptionsParser as Options ( PushOptions (..), )-import Cachix.Client.Push+import Cachix.Client.Push as Push import Cachix.Client.Retry (retryHttp) import Cachix.Client.Secrets import Cachix.Client.Servant@@ -53,9 +53,11 @@ onAttempt = \_ _ -> pass, onUncompressedNARStream = showUploadProgress, onDone = pass,- Cachix.Client.Push.compressionMethod = compressionMethod,- Cachix.Client.Push.compressionLevel = Cachix.Client.OptionsParser.compressionLevel opts,- Cachix.Client.Push.omitDeriver = Cachix.Client.OptionsParser.omitDeriver opts+ Push.compressionMethod = compressionMethod,+ Push.compressionLevel = Options.compressionLevel opts,+ Push.chunkSize = Options.chunkSize opts,+ Push.numConcurrentChunks = Options.numConcurrentChunks opts,+ Push.omitDeriver = Options.omitDeriver opts } where retryText :: RetryStatus -> Text@@ -113,7 +115,7 @@ 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])+ fromMaybe BinaryCache.ZSTD (head $ catMaybes [Options.compressionMethod pushOpts, compressionMethodBackend]) withStore $ \store -> m
src/Cachix/Client/Daemon.hs view
@@ -76,7 +76,7 @@ return $ DaemonEnv {..} --- | Configure and run the daemon. Equivalent to running 'new' and 'run'.+-- | Configure and run the daemon. Equivalent to running 'new' and 'run' together with some signal handling. start :: Env -> DaemonOptions -> PushOptions -> BinaryCacheName -> IO () start daemonEnv daemonOptions daemonPushOptions daemonCacheName = do daemon <- new daemonEnv daemonOptions Nothing daemonPushOptions daemonCacheName@@ -92,60 +92,61 @@ config <- showConfiguration Katip.logFM Katip.InfoS $ Katip.ls $ "Configuration:\n" <> config - let workerCount = Options.numJobs daemonPushOptions- startWorkers pushParams =- Worker.startWorkers- workerCount- (PushManager.pmTaskQueue daemonPushManager)- (liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask pushParams)-- subscriptionManagerThread <-- liftIO $ Async.async $ runSubscriptionManager daemonSubscriptionManager+ Push.withPushParams $ \pushParams -> do+ subscriptionManagerThread <-+ liftIO $ Async.async $ runSubscriptionManager daemonSubscriptionManager - let stopPushManager =- liftIO $ PushManager.stopPushManager daemonPushManager+ let runWorkerTask =+ liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask pushParams+ workersThreads <-+ Worker.startWorkers+ (Options.numJobs daemonPushOptions)+ (PushManager.pmTaskQueue daemonPushManager)+ runWorkerTask - Push.withPushParams $ \pushParams ->- E.bracketOnError (startWorkers pushParams) Worker.stopWorkers $ \workers -> do- flip E.onError stopPushManager $- -- TODO: retry the connection on socket errors- E.bracketOnError (Daemon.openSocket daemonSocketPath) Daemon.closeSocket $ \sock -> do- liftIO $ Socket.listen sock Socket.maxListenQueue+ -- TODO: retry the connection on socket errors+ E.bracketOnError (Daemon.openSocket daemonSocketPath) Daemon.closeSocket $ \sock -> do+ liftIO $ Socket.listen sock Socket.maxListenQueue+ listenThread <- Async.async $ Daemon.listen stop queueJob sock - listenThread <- Async.async $ Daemon.listen stop queueJob sock+ -- Wait for a shutdown signal+ waitForShutdown daemonShutdownLatch - waitForShutdown daemonShutdownLatch+ Katip.logFM Katip.InfoS "Shutting down daemon..." - Katip.logFM Katip.InfoS "Shutting down daemon..."+ -- Stop receiving new push requests+ liftIO $ Socket.shutdown sock Socket.ShutdownReceive `catchAny` \_ -> return () - queuedStorePathCount <- PushManager.runPushManager daemonPushManager PushManager.queuedStorePathCount- when (queuedStorePathCount > 0) $- Katip.logFM Katip.InfoS $- Katip.logStr $- "Remaining store paths: " <> (show queuedStorePathCount :: Text)+ PushManager.runPushManager daemonPushManager $ do+ queuedStorePathCount <- PushManager.queuedStorePathCount+ when (queuedStorePathCount > 0) $+ Katip.logFM Katip.InfoS $+ Katip.logStr $+ "Remaining store paths: " <> (show queuedStorePathCount :: Text) - -- Stop receiving new push requests- liftIO $ Socket.shutdown sock Socket.ShutdownReceive `catchAny` \_ -> return ()+ -- Finish processing remaining push jobs+ liftIO $ PushManager.stopPushManager daemonPushManager - stopPushManager+ -- Gracefully shut down the worker before closing the socket+ Worker.stopWorkers workersThreads - -- Gracefully shutdown the worker *before* closing the socket- Worker.stopWorkers workers+ -- Close all event subscriptions+ liftIO $ stopSubscriptionManager daemonSubscriptionManager+ Async.wait subscriptionManagerThread - liftIO $ stopSubscriptionManager daemonSubscriptionManager- Async.wait subscriptionManagerThread+ -- TODO: say goodbye to all clients waiting for their push to go through+ listenThreadRes <- do+ Async.cancel listenThread+ Async.waitCatch listenThread - -- TODO: say goodbye to all clients waiting for their push to go through- Async.cancel listenThread- res <- Async.waitCatch listenThread- case res of- Right clientSock -> do- -- Wave goodbye to the client that requested the shutdown- liftIO $ Daemon.serverBye clientSock- liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` \_ -> return ()- _ -> return ()+ case listenThreadRes of+ Right clientSock -> do+ -- Wave goodbye to the client that requested the shutdown+ liftIO $ Daemon.serverBye clientSock+ liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` \_ -> return ()+ _ -> return () - return ExitSuccess+ return ExitSuccess stop :: Daemon () stop = asks daemonShutdownLatch >>= initiateShutdown
src/Cachix/Client/Daemon/Log.hs view
@@ -19,7 +19,7 @@ import Katip (renderSeverity) import qualified Katip import qualified Katip.Format.Time as Katip.Format-import Katip.Scribes.Handle (brackets, colorBySeverity)+import Katip.Scribes.Handle (brackets, colorBySeverity, getKeys) import Protolude new :: (MonadIO m) => Katip.Namespace -> Maybe Handle -> LogLevel -> m Logger@@ -66,12 +66,14 @@ Error -> Katip.ErrorS conciseBracketFormat :: (Katip.LogItem a) => Katip.ItemFormatter a-conciseBracketFormat withColor _verbosity Katip.Item {..} =+conciseBracketFormat withColor verbosity Katip.Item {..} = brackets nowStr <> brackets (fromText (renderSeverity' _itemSeverity))+ <> mconcat ks <> fromText " " <> Katip.unLogStr _itemMessage where nowStr = fromText (Katip.Format.formatAsLogTime _itemTime)+ ks = map brackets $ getKeys verbosity _itemPayload renderSeverity' severity = colorBySeverity withColor severity (renderSeverity severity)
src/Cachix/Client/Daemon/PushManager.hs view
@@ -46,7 +46,6 @@ import qualified Conduit as C import Control.Concurrent.STM.TBMQueue import Control.Concurrent.STM.TVar-import qualified Control.Exception.Safe as Safe import qualified Control.Monad.Catch as E import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Control.Retry (RetryStatus)@@ -182,13 +181,12 @@ resolvePushJob :: Protocol.PushRequestId -> PushJob.ResolvedClosure FilePath -> PushManager () resolvePushJob pushId closure = do- timestamp <- liftIO getCurrentTime+ Katip.logLocM Katip.DebugS $ Katip.ls $ showClosureStats closure + timestamp <- liftIO getCurrentTime _ <- modifyPushJob pushId $ PushJob.populateQueue closure timestamp withPushJob pushId $ \pushJob -> do- Katip.logLocM Katip.DebugS $ Katip.ls $ showClosureStats closure- pushStarted pushJob -- Create STM action for each path and then run everything atomically queueStorePaths pushId $ Set.toList (PushJob.rcMissingPaths closure)@@ -212,28 +210,53 @@ handleTask :: PushParams PushManager () -> Task -> PushManager () handleTask pushParams task = do case task of- ResolveClosure pushId -> do+ ResolveClosure pushId ->+ runResolveClosureTask pushParams pushId+ PushStorePath filePath ->+ runPushStorePathTask pushParams filePath++runResolveClosureTask :: PushParams PushManager () -> Protocol.PushRequestId -> PushManager ()+runResolveClosureTask pushParams pushId =+ resolveClosure `withException` failJob+ where+ failJob :: SomeException -> PushManager ()+ failJob err = do+ failPushJob pushId++ Katip.katipAddContext (Katip.sl "error" (displayException err)) $+ Katip.logLocM Katip.ErrorS $+ Katip.ls $+ "Failed to resolve closure for push job " <> (show pushId :: Text)++ resolveClosure = do Katip.logLocM Katip.DebugS $ Katip.ls $ "Resolving closure for push job " <> (show pushId :: Text) - withPushJob pushId $ \pushJob ->- E.onException (failPushJob pushId) $ do- let sps = Protocol.storePaths (pushRequest pushJob)- store = pushParamsStore pushParams- normalized <- mapM (normalizeStorePath store) sps- (allStorePaths, missingStorePaths) <- getMissingPathsForClosure pushParams (catMaybes normalized)- storePathsToPush <- pushOnClosureAttempt pushParams allStorePaths missingStorePaths+ withPushJob pushId $ \pushJob -> do+ let sps = Protocol.storePaths (pushRequest pushJob)+ store = pushParamsStore pushParams+ normalized <- mapM (normalizeStorePath store) sps+ (allStorePaths, missingStorePaths) <- getMissingPathsForClosure pushParams (catMaybes normalized)+ storePathsToPush <- pushOnClosureAttempt pushParams allStorePaths missingStorePaths - resolvedClosure <- do- allPaths <- liftIO $ mapM (storeToFilePath store) allStorePaths- pathsToPush <- liftIO $ mapM (storeToFilePath store) storePathsToPush- return $- PushJob.ResolvedClosure- { rcAllPaths = Set.fromList allPaths,- rcMissingPaths = Set.fromList pathsToPush- }+ resolvedClosure <- do+ allPaths <- liftIO $ mapM (storeToFilePath store) allStorePaths+ pathsToPush <- liftIO $ mapM (storeToFilePath store) storePathsToPush+ return $+ PushJob.ResolvedClosure+ { rcAllPaths = Set.fromList allPaths,+ rcMissingPaths = Set.fromList pathsToPush+ } - resolvePushJob pushId resolvedClosure- PushStorePath filePath -> do+ resolvePushJob pushId resolvedClosure++runPushStorePathTask :: PushParams PushManager () -> FilePath -> PushManager ()+runPushStorePathTask pushParams filePath = do+ pushStorePath `withException` failStorePath+ where+ failStorePath =+ pushStorePathFailed filePath . toS . displayException++ pushStorePath = do qs <- asks pmTaskSemaphore E.bracket_ (QSem.waitQSem qs) (QSem.signalQSem qs) $ do Katip.logLocM Katip.DebugS $ Katip.ls $ "Pushing store path " <> filePath@@ -241,8 +264,7 @@ let store = pushParamsStore pushParams storePath <- liftIO $ parseStorePath store (toS filePath) - retryAll (uploadStorePath pushParams storePath)- `Safe.catchAny` (pushStorePathFailed filePath . toS . displayException)+ retryAll $ uploadStorePath pushParams storePath newPushStrategy :: Store ->@@ -298,6 +320,8 @@ onDone = onDone, Client.Push.compressionMethod = compressionMethod, Client.Push.compressionLevel = Client.OptionsParser.compressionLevel opts,+ Client.Push.chunkSize = Client.OptionsParser.chunkSize opts,+ Client.Push.numConcurrentChunks = Client.OptionsParser.numConcurrentChunks opts, Client.Push.omitDeriver = Client.OptionsParser.omitDeriver opts } @@ -385,8 +409,12 @@ fp <- liftIO $ storePathToPath store storePath pure $ toS fp +-- | Canonicalize and validate a store path normalizeStorePath :: (MonadIO m) => Store -> FilePath -> m (Maybe StorePath) normalizeStorePath store fp = liftIO $ runMaybeT $ do storePath <- MaybeT $ followLinksToStorePath store (encodeUtf8 $ T.pack fp) MaybeT $ filterInvalidStorePath store storePath++withException :: (E.MonadCatch m) => m a -> (SomeException -> m a) -> m a+withException action handler = action `E.catchAll` (\e -> handler e >> E.throwM e)
src/Cachix/Client/Daemon/Types/PushManager.hs view
@@ -33,15 +33,20 @@ type PushJobStore = TVar (HashMap Protocol.PushRequestId PushJob) +-- TODO: a lot of the logic surrounding deduping, search, and job tracking could be replaced by sqlite.+-- sqlite can run in-memory if we don't need persistence.+-- If we do, then we can we get stop/resume for free. data PushManagerEnv = PushManagerEnv- { pmPushJobs :: PushJobStore,- -- | A mapping of store paths to to push requests.- -- Use to prevent duplicate pushes and track with store paths are referenced by push requests.+ { -- | A store of push jobs indexed by a PushRequestId.+ pmPushJobs :: PushJobStore,+ -- | A mapping of store paths to push requests.+ -- Used to prevent duplicate pushes and track which store paths are referenced by which push requests. pmStorePathReferences :: TVar (HashMap FilePath [Protocol.PushRequestId]), -- | FIFO queue of push tasks. pmTaskQueue :: TBMQueue Task,+ -- | A semaphore to control task concurrency. pmTaskSemaphore :: QSem,- -- | Callback for push events.+ -- | A callback for push events. pmOnPushEvent :: OnPushEvent, pmLogger :: Logger }
src/Cachix/Client/OptionsParser.hs view
@@ -5,12 +5,26 @@ DaemonCommand (..), DaemonOptions (..), PushArguments (..),++ -- * Push options PushOptions (..), defaultPushOptions,+ defaultCompressionMethod,+ defaultCompressionLevel,+ defaultNumConcurrentChunks,+ defaultChunkSize,+ defaultNumJobs,+ defaultOmitDeriver,++ -- * Pin options PinOptions (..),++ -- * Global options+ Flags (..),++ -- * Misc BinaryCacheName, getOpts,- Flags (..), ) where @@ -22,6 +36,7 @@ import Cachix.Types.BinaryCache (BinaryCacheName) import qualified Cachix.Types.BinaryCache as BinaryCache import Cachix.Types.PinCreate (Keep (..))+import Data.Conduit.ByteString (ChunkSize) import qualified Data.Text as T import Options.Applicative import Protolude hiding (toS)@@ -107,9 +122,24 @@ deriving (Show) data PushOptions = PushOptions- { compressionLevel :: Int,+ { -- | The compression level to use.+ compressionLevel :: Int,+ -- | The compression method to use.+ -- Default value taken rom the cache settings. compressionMethod :: Maybe BinaryCache.CompressionMethod,+ -- | The size of each uploaded part.+ --+ -- Common values for S3 are powers of 2 (8MiB, 16MiB, and so on), but this doesn't appear to be a requirement for S3 (unlike Glacier).+ -- The range is from 5MiB to 5GiB.+ --+ -- Lower values will increase HTTP overhead, higher values require more memory to preload each part.+ chunkSize :: Int,+ -- | The number of chunks to upload concurrently.+ -- The total memory usage is numJobs * numConcurrentChunks * chunkSize+ numConcurrentChunks :: Int,+ -- | The number of store paths to process concurrently. numJobs :: Int,+ -- | Omit the derivation from the store path metadata. omitDeriver :: Bool } deriving (Show)@@ -117,9 +147,21 @@ defaultCompressionLevel :: Int defaultCompressionLevel = 2 -defaultCompressionMethod :: Maybe BinaryCache.CompressionMethod-defaultCompressionMethod = Nothing+defaultCompressionMethod :: BinaryCache.CompressionMethod+defaultCompressionMethod = BinaryCache.ZSTD +defaultNumConcurrentChunks :: Int+defaultNumConcurrentChunks = 4++defaultChunkSize :: ChunkSize+defaultChunkSize = 32 * 1024 * 1024 -- 32MiB++minChunkSize :: ChunkSize+minChunkSize = 5 * 1024 * 1024 -- 5MiB++maxChunkSize :: ChunkSize+maxChunkSize = 5 * 1024 * 1024 * 1024 -- 5GiB+ defaultNumJobs :: Int defaultNumJobs = 8 @@ -130,7 +172,9 @@ defaultPushOptions = PushOptions { compressionLevel = defaultCompressionLevel,- compressionMethod = defaultCompressionMethod,+ compressionMethod = Nothing,+ chunkSize = defaultChunkSize,+ numConcurrentChunks = defaultNumConcurrentChunks, numJobs = defaultNumJobs, omitDeriver = defaultOmitDeriver }@@ -178,6 +222,9 @@ Right a -> Right $ Just a Left b -> Left $ toS b else Left $ "Compression method " <> show method <> " not expected. Use xz or zstd."+ validatedChunkSize c =+ c <$ unless (c >= minChunkSize && c <= maxChunkSize) (readerError $ "value " <> show c <> " not in expected range: " <> prettyChunkRange)+ prettyChunkRange = "[" <> show minChunkSize <> ".." <> show maxChunkSize <> "]" pushOptions :: Parser PushOptions pushOptions = PushOptions@@ -197,13 +244,32 @@ <> short 'm' <> metavar "xz | zstd" <> help- "The compression method to use. Supported methods: xz | zstd. Defaults to zstd."- <> value defaultCompressionMethod+ "The compression method to use. Overrides the preferred compression method advertised by the cache. Supported methods: xz | zstd. Defaults to zstd."+ <> value Nothing ) <*> option+ (auto >>= validatedChunkSize)+ ( long "chunk-size"+ <> short 's'+ <> metavar prettyChunkRange+ <> help "The size of each uploaded part in bytes. The supported range is from 5MiB to 5GiB."+ <> showDefault+ <> value defaultChunkSize+ )+ <*> option auto+ ( long "num-concurrent-chunks"+ <> short 'n'+ <> metavar "INT"+ <> help "The number of chunks to upload concurrently. The total memory usage is jobs * num-concurrent-chunks * chunk-size."+ <> showDefault+ <> value defaultNumConcurrentChunks+ )+ <*> option+ auto ( long "jobs" <> short 'j'+ <> metavar "INT" <> help "The number of threads to use when pushing store paths." <> showDefault <> value defaultNumJobs
src/Cachix/Client/Push.hs view
@@ -23,7 +23,7 @@ newPathInfoFromNarInfo, -- * Streaming upload- Push.S3.UploadResult (..),+ Push.S3.UploadMultipartResult (..), UploadNarDetails (..), streamUploadNar, streamCopy,@@ -61,6 +61,7 @@ import Crypto.Sign.Ed25519 import qualified Data.ByteString.Base64 as B64 import Data.Conduit+import Data.Conduit.ByteString (ChunkSize) import qualified Data.Conduit.Lzma as Lzma (compress) import qualified Data.Conduit.Zstd as Zstd (compress) import Data.IORef@@ -134,8 +135,15 @@ onError :: ClientError -> m r, -- | An action to run after the path is pushed successfully. onDone :: m r,+ -- | The compression method to use. compressionMethod :: BinaryCache.CompressionMethod,+ -- | The compression level to use. compressionLevel :: Int,+ -- | The chunk size to use.+ chunkSize :: ChunkSize,+ -- | The number of chunks to upload concurrently.+ numConcurrentChunks :: Int,+ -- | Whether to mit the deriver from the narinfo. omitDeriver :: Bool } @@ -243,10 +251,10 @@ completeNarUpload :: (MonadUnliftIO m) => PushParams n r ->- Push.S3.UploadResult ->+ Push.S3.UploadMultipartResult -> Api.NarInfoCreate -> m ()-completeNarUpload pushParams Push.S3.UploadResult {..} nic = do+completeNarUpload pushParams Push.S3.UploadMultipartResult {..} nic = do let cacheName = pushParamsName pushParams authToken = getCacheAuthToken (pushParamsSecret pushParams) clientEnv = pushParamsClientEnv pushParams@@ -324,7 +332,7 @@ StorePath -> Int64 -> RetryStatus ->- ConduitT ByteString Void (ResourceT m) (Either ClientError (Push.S3.UploadResult, UploadNarDetails))+ ConduitT ByteString Void (ResourceT m) (Either ClientError (Push.S3.UploadMultipartResult, UploadNarDetails)) streamUploadNar pushParams storePath storePathSize retrystatus = do let cacheName = pushParamsName pushParams authToken = getCacheAuthToken (pushParamsSecret pushParams)@@ -335,6 +343,13 @@ BinaryCache.XZ -> defaultWithXzipCompressorWithLevel (compressionLevel strategy) BinaryCache.ZSTD -> defaultWithZstdCompressorWithLevel (compressionLevel strategy) + let uploadOptions =+ Push.S3.UploadMultipartOptions+ { Push.S3.numConcurrentChunks = numConcurrentChunks strategy,+ Push.S3.chunkSize = chunkSize strategy,+ Push.S3.compressionMethod = compressionMethod strategy+ }+ narSizeRef <- liftIO $ newIORef 0 fileSizeRef <- liftIO $ newIORef 0 narHashRef <- liftIO $ newIORef ("" :: ByteString)@@ -348,7 +363,7 @@ .| compressor .| passthroughSizeSink fileSizeRef .| passthroughHashSinkB16 fileHashRef- .| Push.S3.streamUpload clientEnv authToken cacheName (compressionMethod strategy)+ .| Push.S3.uploadMultipart clientEnv authToken cacheName uploadOptions for result $ \uploadResult -> liftIO $ do narSize <- readIORef narSizeRef@@ -375,13 +390,21 @@ Int64 -> RetryStatus -> BinaryCache.CompressionMethod ->- ConduitT ByteString Void (ResourceT m) (Either ClientError (Push.S3.UploadResult, UploadNarDetails))+ ConduitT ByteString Void (ResourceT m) (Either ClientError (Push.S3.UploadMultipartResult, UploadNarDetails)) streamCopy pushParams storePath claimedFileSize retrystatus compressionMethod = do let cacheName = pushParamsName pushParams authToken = getCacheAuthToken (pushParamsSecret pushParams) clientEnv = pushParamsClientEnv pushParams strategy = pushParamsStrategy pushParams storePath + let uploadOptions =+ Push.S3.UploadMultipartOptions+ { Push.S3.numConcurrentChunks = numConcurrentChunks strategy,+ Push.S3.chunkSize = chunkSize strategy,+ -- TODO: why is this not part of the strategy?+ Push.S3.compressionMethod = compressionMethod+ }+ fileSizeRef <- liftIO $ newIORef 0 fileHashRef <- liftIO $ newIORef ("" :: ByteString) @@ -390,7 +413,7 @@ .| onUncompressedNARStream strategy retrystatus claimedFileSize .| passthroughSizeSink fileSizeRef .| passthroughHashSinkB16 fileHashRef- .| Push.S3.streamUpload clientEnv authToken cacheName compressionMethod+ .| Push.S3.uploadMultipart clientEnv authToken cacheName uploadOptions for result $ \uploadResult -> liftIO $ do fileSize <- readIORef fileSizeRef
src/Cachix/Client/Push/S3.hs view
@@ -2,7 +2,12 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE Rank2Types #-} -module Cachix.Client.Push.S3 where+module Cachix.Client.Push.S3+ ( UploadMultipartResult (..),+ UploadMultipartOptions (..),+ uploadMultipart,+ )+where import qualified Cachix.API as API import Cachix.API.Error@@ -31,60 +36,52 @@ 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+data UploadMultipartOptions = UploadMultipartOptions+ { numConcurrentChunks :: Int,+ chunkSize :: ChunkSize,+ compressionMethod :: CompressionMethod+ } -data UploadResult = UploadResult+data UploadMultipartResult = UploadMultipartResult { urNarId :: UUID, urUploadId :: Text, urParts :: Maybe (NonEmpty Multipart.CompletedPart) } deriving stock (Eq, Show) -streamUpload ::+uploadMultipart :: forall m. (MonadUnliftIO m, MonadResource m) => ClientEnv -> Token -> Text ->- CompressionMethod ->+ UploadMultipartOptions -> ConduitT ByteString Void m- (Either ClientError UploadResult)-streamUpload env authToken cacheName compressionMethod = do+ (Either ClientError UploadMultipartResult)+uploadMultipart env authToken cacheName options = do createMultipartUpload >>= \case Left err -> return $ Left err Right (Multipart.CreateMultipartUploadResponse {narId, uploadId}) -> do handleC (abortMultipartUpload narId uploadId) $- chunkStream (Just chunkSize)- .| concurrentMapM_ concurrentParts outputBufferSize (uploadPart narId uploadId)+ chunkStream (Just (chunkSize options))+ .| concurrentMapM_ (numConcurrentChunks options) outputBufferSize (uploadPart narId uploadId) .| completeMultipartUpload narId uploadId where+ -- 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 = 2 * numConcurrentChunks options+ manager = Client.manager env createMultipartUpload :: ConduitT ByteString Void m (Either ClientError Multipart.CreateMultipartUploadResponse) createMultipartUpload = do- let createNarRequest = API.createNar cachixClient authToken cacheName (Just compressionMethod)+ let createNarRequest = API.createNar cachixClient authToken cacheName (Just (compressionMethod options)) liftIO $ retryHttp $ runClientM createNarRequest env uploadPart :: UUID -> Text -> (Int, ByteString) -> m (Maybe Multipart.CompletedPart)@@ -117,7 +114,7 @@ parts <- CC.sinkList return $ Right $- UploadResult+ UploadMultipartResult { urNarId = narId, urUploadId = uploadId, urParts = sequenceA (NonEmpty.fromList parts)