diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. 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.7.1 +version: 1.7.2 synopsis: Command-line client for Nix binary cache hosting https://cachix.org diff --git a/src/Cachix/Client/Commands.hs b/src/Cachix/Client/Commands.hs --- a/src/Cachix/Client/Commands.hs +++ b/src/Cachix/Client/Commands.hs @@ -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 $ diff --git a/src/Cachix/Client/Commands/Push.hs b/src/Cachix/Client/Commands/Push.hs --- a/src/Cachix/Client/Commands/Push.hs +++ b/src/Cachix/Client/Commands/Push.hs @@ -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 diff --git a/src/Cachix/Client/Daemon.hs b/src/Cachix/Client/Daemon.hs --- a/src/Cachix/Client/Daemon.hs +++ b/src/Cachix/Client/Daemon.hs @@ -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 diff --git a/src/Cachix/Client/Daemon/Log.hs b/src/Cachix/Client/Daemon/Log.hs --- a/src/Cachix/Client/Daemon/Log.hs +++ b/src/Cachix/Client/Daemon/Log.hs @@ -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) diff --git a/src/Cachix/Client/Daemon/PushManager.hs b/src/Cachix/Client/Daemon/PushManager.hs --- a/src/Cachix/Client/Daemon/PushManager.hs +++ b/src/Cachix/Client/Daemon/PushManager.hs @@ -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) diff --git a/src/Cachix/Client/Daemon/Types/PushManager.hs b/src/Cachix/Client/Daemon/Types/PushManager.hs --- a/src/Cachix/Client/Daemon/Types/PushManager.hs +++ b/src/Cachix/Client/Daemon/Types/PushManager.hs @@ -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 } diff --git a/src/Cachix/Client/OptionsParser.hs b/src/Cachix/Client/OptionsParser.hs --- a/src/Cachix/Client/OptionsParser.hs +++ b/src/Cachix/Client/OptionsParser.hs @@ -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 diff --git a/src/Cachix/Client/Push.hs b/src/Cachix/Client/Push.hs --- a/src/Cachix/Client/Push.hs +++ b/src/Cachix/Client/Push.hs @@ -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 diff --git a/src/Cachix/Client/Push/S3.hs b/src/Cachix/Client/Push/S3.hs --- a/src/Cachix/Client/Push/S3.hs +++ b/src/Cachix/Client/Push/S3.hs @@ -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)