packages feed

cachix 0.5.1 → 0.6.0

raw patch · 17 files changed

+343/−158 lines, 17 filesdep +concurrent-extradep +stmdep ~dhall

Dependencies added: concurrent-extra, stm

Dependency ranges changed: dhall

Files

CHANGELOG.md view
@@ -7,6 +7,27 @@  ## Unreleased +## [0.6.0] - 2021-01-08++### Changed++- Watching nix store doesn't push .drv files anymore+- `cachix push -w` has been moved to `cachix watch-store`+- `cachix create` has been removed+- Retries now take a multiple of seconds instead of multiple of 100ms++### Added++- watch-exec allows to run a command and push all new store paths added meanwhile+- GHC 8.10 support++### Fixed++- Watching /nix/store now uses queue to bulk query what is missing in binarycache and+  a queue for pushing+- Instructions for NixOS trusted users were inaccurate+- Retry fetching binary cache+ ## [0.5.1] - 2020-11-09  ### Fixed
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version: 0.5.1+version: 0.6.0 license:            Apache-2.0 license-file:       LICENSE copyright:          2018 Domen Kožar@@ -55,35 +55,32 @@     Cachix.Client.NixVersion     Cachix.Client.OptionsParser     Cachix.Client.Push+    Cachix.Client.PushQueue+    Cachix.Client.Retry     Cachix.Client.Secrets     Cachix.Client.Servant     Cachix.Client.Store     Cachix.Client.Store.Context     Cachix.Client.URI+    Cachix.Client.WatchStore     System.Nix.Base32 -  cc-options:        -Wall-  pkgconfig-depends: nix-store ==2.0 || >2.0, nix-main ==2.0 || >2.0   hs-source-dirs:    src   other-modules:     Paths_cachix   autogen-modules:   Paths_cachix-  extra-libraries:-    stdc++-    boost_context--  include-dirs:      cbits   build-depends:     , async     , base                  >=4.7     && <5     , base64-bytestring     , bytestring     , cachix-api+    , concurrent-extra     , conduit               >=1.3.0     , conduit-extra     , containers     , cookie     , cryptonite-    , dhall+    , dhall                 >=1.28.0     , directory     , ed25519     , filepath@@ -112,13 +109,20 @@     , servant-client        >=0.16     , servant-client-core   >=0.16     , servant-conduit+    , stm     , text     , unix     , uri-bytestring     , vector     , versions -  ghc-options:       -optc=-std=c++17+  extra-libraries:+    stdc+++    boost_context++  include-dirs:      cbits+  pkgconfig-depends: nix-store ==2.0 || >2.0, nix-main ==2.0 || >2.0+  ghc-options:       -optcxx-std=c++17    if os(osx)     ghc-options: -pgmc=clang++
src/Cachix/Client.hs view
@@ -14,8 +14,9 @@   env <- mkEnv cachixoptions   case command of     AuthToken token -> Commands.authtoken env token-    Create name -> Commands.create env name     GenerateKeypair name -> Commands.generateKeypair env name     Push pushArgs -> Commands.push env pushArgs+    WatchStore watchArgs name -> Commands.watchStore env watchArgs name+    WatchExec pushArgs name cmd args -> Commands.watchExec env pushArgs name cmd args     Use name useOptions -> Commands.use env name useOptions     Version -> putText cachixVersion
src/Cachix/Client/Commands.hs view
@@ -5,9 +5,10 @@  module Cachix.Client.Commands   ( authtoken,-    create,     generateKeypair,     push,+    watchStore,+    watchExec,     use,   ) where@@ -33,29 +34,31 @@     PushOptions (..),   ) import Cachix.Client.Push+import Cachix.Client.Retry (retryAll) import Cachix.Client.Secrets   ( SigningKey (SigningKey),     exportSigningKey,   ) import Cachix.Client.Servant-import Cachix.Client.Store (Store)+import qualified Cachix.Client.WatchStore as WatchStore import qualified Cachix.Types.SigningKeyCreate as SigningKeyCreate-import Control.Exception.Safe (handle, throwM)+import Control.Exception.Safe (throwM) import Control.Retry (RetryStatus (rsIterNumber))-import Crypto.Sign.Ed25519+import Crypto.Sign.Ed25519 (PublicKey (PublicKey), createKeypair) import qualified Data.ByteString.Base64 as B64-import Data.List (isSuffixOf) import Data.String.Here import qualified Data.Text as T import Network.HTTP.Types (status401, status404)-import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.API (NoContent) import Servant.Auth.Client import Servant.Client.Streaming import Servant.Conduit () import System.Directory (doesFileExist)-import System.FSNotify import System.IO (hIsTerminalDevice)+import qualified System.Posix.Signals as Signals+import qualified System.Process  authtoken :: Env -> Text -> IO () authtoken env token = do@@ -64,10 +67,6 @@     Just cfg -> Config.setAuthToken cfg $ Token (toS token)     Nothing -> mkConfig token -create :: Env -> Text -> IO ()-create _ _ =-  throwIO $ DeprecatedCommand "Create command has been deprecated. Please visit https://app.cachix.org to create a binary cache."- generateKeypair :: Env -> Text -> IO () generateKeypair env name = do   cachixAuthToken <- Config.getAuthTokenRequired (config env)@@ -77,8 +76,9 @@       bcc = BinaryCacheConfig name signingKey   -- we first validate if key can be added to the binary cache   (_ :: NoContent) <--    escalate <=< (`runClientM` clientenv env) $-      API.createKey cachixClient cachixAuthToken name signingKeyCreate+    escalate <=< retryAll $ \_ ->+      (`runClientM` clientenv env) $+        API.createKey cachixClient cachixAuthToken name signingKeyCreate   -- if key was successfully added, write it to the config   -- TODO: warn if binary cache with the same key already exists   let cfg = case config env of@@ -120,7 +120,7 @@ use env name useOptions = do   cachixAuthToken <- Config.getAuthTokenOptional (config env)   -- 1. get cache public key-  res <- (`runClientM` clientenv env) $ API.getCache cachixClient cachixAuthToken name+  res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient cachixAuthToken name   case res of     Left err       | isErr err status401 && isJust (config env) -> throwM $ accessDeniedBinaryCache name@@ -142,7 +142,6 @@       InstallationMode.addBinaryCache (config env) binaryCache useOptions $         InstallationMode.getInstallationMode nixEnv useOptions --- TODO: lots of room for performance improvements push :: Env -> PushArguments -> IO () push env (PushPaths opts name cliPaths) = do   hasStdin <- not <$> hIsTerminalDevice stdin@@ -156,51 +155,34 @@       -- that may or may not be written to by the caller.       -- This is somewhat like the behavior of `cat` for example.       (_, paths) -> return paths-  cacheSecret <- findPushSecret (config env) name-  store <- wait (storeAsync env)+  pushParams <- getPushParams env opts name   void $     pushClosure       (mapConcurrentlyBounded (numJobs opts))-      (clientenv env)-      store-      PushCache-        { pushCacheName = name,-          pushCacheSecret = cacheSecret-        }-      (pushStrategy env opts name)+      pushParams       inputStorePaths   putText "All done."-push env (PushWatchStore opts name) = withManager $ \mgr -> do-  store <- wait (storeAsync env)-  pushSecret <- findPushSecret (config env) name-  _ <- watchDir mgr "/nix/store" filterF (action store pushSecret)-  putText "Watching /nix/store for new paths ..."-  forever $ threadDelay (1000 * 1000)+push _ _ = do+  throwIO $ DeprecatedCommand "DEPRECATED: cachix watch-store has replaced cachix push --watch-store."++watchStore :: Env -> PushOptions -> Text -> IO ()+watchStore env opts name = do+  pushParams <- getPushParams env opts name+  WatchStore.startWorkers (numJobs opts) pushParams++watchExec :: Env -> PushOptions -> Text -> Text -> [Text] -> IO ()+watchExec env pushOpts name cmd args = do+  pushParams <- getPushParams env pushOpts name+  let watch = WatchStore.startWorkers (numJobs pushOpts) pushParams+  (_, exitCode) <- concurrently watch run+  exitWith exitCode   where-    action :: Store -> PushSecret -> Action-    action store pushSecret (Removed fp _ _) =-      let storePath = toS $ dropEnd 5 fp-       in Control.Exception.Safe.handle (logErr fp) $ void $-            pushClosure-              (mapConcurrentlyBounded (numJobs opts))-              (clientenv env)-              store-              ( PushCache-                  { pushCacheName = name,-                    pushCacheSecret = pushSecret-                  }-              )-              (pushStrategy env opts name)-              [storePath]-    action _ _ _ = return ()-    logErr :: FilePath -> SomeException -> IO ()-    logErr fp e = hPutStrLn stderr $ "Exception occured while pushing " <> fp <> ": " <> show e-    filterF :: ActionPredicate-    filterF (Removed fp _ _)-      | ".lock" `isSuffixOf` fp = True-    filterF _ = False-    dropEnd :: Int -> [a] -> [a]-    dropEnd index xs = take (length xs - index) xs+    process = System.Process.proc (toS cmd) (toS <$> args)+    run = do+      (_, _, _, processHandle) <- System.Process.createProcess process+      exitCode <- System.Process.waitForProcess processHandle+      Signals.raiseSignal Signals.sigINT+      return exitCode  retryText :: RetryStatus -> Text retryText retrystatus =@@ -224,3 +206,16 @@       withXzipCompressor = defaultWithXzipCompressorWithLevel (compressionLevel opts),       Cachix.Client.Push.omitDeriver = Cachix.Client.OptionsParser.omitDeriver opts     }++getPushParams :: Env -> PushOptions -> Text -> IO (PushParams IO ())+getPushParams env pushOpts name = do+  pushSecret <- findPushSecret (config env) name+  store <- wait (storeAsync env)+  return $+    PushParams+      { pushParamsName = name,+        pushParamsSecret = pushSecret,+        pushParamsClientEnv = clientenv env,+        pushParamsStrategy = pushStrategy env pushOpts name,+        pushParamsStore = store+      }
src/Cachix/Client/Config.hs view
@@ -22,7 +22,8 @@ import Data.String.Here import Dhall hiding (Text) import Dhall.Pretty (prettyExpr)-import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.Auth.Client import System.Directory   ( XdgDirectory (..),
src/Cachix/Client/Config/Orphans.hs view
@@ -1,5 +1,4 @@ {-# OPTIONS_GHC -Wno-orphans #-}-{-# LANGUAGE CPP #-}  module Cachix.Client.Config.Orphans   (@@ -8,30 +7,18 @@  import Dhall hiding (Text) import Dhall.Core (Chunks (..), Expr (..))-import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.Auth.Client -#if MIN_VERSION_dhall(1,28,0) instance FromDhall Token where-  autoWith _ = Decoder-#else-instance Interpret Token where-  autoWith _ = Type-#endif-    { extract = ex,-      expected = Text-    }+  autoWith _ = strictText { extract = ex }     where       ex (TextLit (Chunks [] t)) = pure (Token (toS t))       ex _ = panic "Unexpected Dhall value. Did it typecheck?" -#if MIN_VERSION_dhall(1,28,0) instance ToDhall Token where   injectWith _ = Encoder-#else-instance Inject Token where-  injectWith _ = InputType-#endif     { embed = TextLit . Chunks [] . toS . getToken,       declared = Text     }
src/Cachix/Client/Env.hs view
@@ -20,7 +20,8 @@ import Network.HTTP.Client.TLS (newTlsManagerWith, tlsManagerSettings) import Network.HTTP.Simple (setRequestHeader) import Paths_cachix (version)-import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.Client (ClientEnv, mkClientEnv) import System.Directory (canonicalizePath)
src/Cachix/Client/InstallationMode.hs view
@@ -92,7 +92,7 @@ b) Add the following to your configuration.nix to add your user as trusted     and then try again: -  trustedUsers = [ "root" "${user}" ];+  nix.trustedUsers = [ "root" "${user}" ];      |] addBinaryCache _ _ _ UntrustedRequiresSudo = do
src/Cachix/Client/NetRc.hs view
@@ -11,7 +11,8 @@ import Data.List (nubBy) import qualified Data.Text as T import Network.NetRc-import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.Auth.Client (Token, getToken) import System.Directory (createDirectoryIfMissing, doesFileExist) import System.FilePath (takeDirectory)
src/Cachix/Client/OptionsParser.hs view
@@ -12,7 +12,8 @@ import qualified Cachix.Client.InstallationMode as InstallationMode import Cachix.Client.URI (defaultCachixURI) import Options.Applicative-import Protolude hiding (option)+import Protolude hiding (option, toS)+import Protolude.Conv import URI.ByteString   ( Absolute,     URIRef,@@ -62,9 +63,10 @@  data CachixCommand   = AuthToken Text-  | Create BinaryCacheName   | GenerateKeypair BinaryCacheName   | Push PushArguments+  | WatchStore PushOptions Text+  | WatchExec PushOptions Text Text [Text]   | Use BinaryCacheName InstallationMode.UseOptions   | Version   deriving (Show)@@ -85,15 +87,15 @@ parserCachixCommand :: Parser CachixCommand parserCachixCommand =   subparser $-    command "authtoken" (infoH authtoken (progDesc "Configure authentication token for communication to cachix.org API"))-      <> command "create" (infoH create (progDesc "DEPRECATED: Go to https://cachix.org instead"))-      <> command "generate-keypair" (infoH generateKeypair (progDesc "Generate keypair for a binary cache"))+    command "authtoken" (infoH authtoken (progDesc "Configure authentication token for communication to HTTP API"))+      <> command "generate-keypair" (infoH generateKeypair (progDesc "Generate signing key pair for a binary cache"))       <> command "push" (infoH push (progDesc "Upload Nix store paths to a binary cache"))-      <> command "use" (infoH use (progDesc "Configure a binary cache by writing nix.conf and netrc files."))+      <> command "watch-exec" (infoH watchExec (progDesc "Run a command while it's running watch /nix/store for newly added store paths and upload them to a binary cache"))+      <> command "watch-store" (infoH watchStore (progDesc "Indefinitely watch /nix/store for newly added store paths and upload them to a binary cache"))+      <> command "use" (infoH use (progDesc "Configure a binary cache by writing nix.conf and netrc files"))   where     nameArg = strArgument (metavar "CACHE-NAME")-    authtoken = AuthToken <$> strArgument (metavar "AUTHTOKEN")-    create = Create <$> nameArg+    authtoken = AuthToken <$> strArgument (metavar "AUTH-TOKEN")     generateKeypair = GenerateKeypair <$> nameArg     validatedLevel l =       l <$ unless (l `elem` [0 .. 9]) (readerError $ "value " <> show l <> " not in expected range: [0..9]")@@ -124,13 +126,15 @@     pushPaths =       (\paths opts cache -> PushPaths opts cache paths)         <$> many (strArgument (metavar "PATHS..."))+    watchExec = WatchExec <$> pushOptions <*> nameArg <*> strArgument (metavar "CMD") <*> many (strArgument (metavar "-- ARGS"))+    watchStore = WatchStore <$> pushOptions <*> nameArg     pushWatchStore =       (\() opts cache -> PushWatchStore opts cache)         <$> flag'           ()           ( long "watch-store"               <> short 'w'-              <> help "Run in daemon mode and push store paths as they are added to /nix/store"+              <> help "DEPRECATED: use watch-store command instead."           )     use =       Use <$> nameArg@@ -180,8 +184,8 @@ desc :: InfoMod a desc =   fullDesc-    <> progDesc "Sign into https://cachix.org to get started."-    <> header "cachix.org command interface"+    <> progDesc "To get started log in to https://app.cachix.org"+    <> header "https://cachix.org command line interface"  -- TODO: usage footer infoH :: Parser a -> InfoMod a -> ParserInfo a
src/Cachix/Client/Push.hs view
@@ -8,7 +8,8 @@ module Cachix.Client.Push   ( -- * Pushing a single path     pushSingleStorePath,-    PushCache (..),+    uploadStorePath,+    PushParams (..),     PushSecret (..),     PushStrategy (..),     defaultWithXzipCompressor,@@ -17,6 +18,7 @@      -- * Pushing a closure of store paths     pushClosure,+    getMissingPathsForClosure,     mapConcurrentlyBounded,   ) where@@ -26,6 +28,7 @@ import Cachix.API.Signing (fingerprint, passthroughHashSink, passthroughHashSinkB16, passthroughSizeSink) import qualified Cachix.Client.Config as Config import Cachix.Client.Exception (CachixException (..))+import Cachix.Client.Retry (retryAll) import Cachix.Client.Secrets import Cachix.Client.Servant import Cachix.Client.Store (Store)@@ -37,7 +40,7 @@ import qualified Control.Concurrent.QSem as QSem import Control.Exception.Safe (MonadMask, throwM) import Control.Monad.Trans.Resource (ResourceT)-import Control.Retry (RetryPolicy, RetryStatus, exponentialBackoff, limitRetries, recoverAll)+import Control.Retry (RetryStatus) import Crypto.Sign.Ed25519 import qualified Data.ByteString.Base64 as B64 import Data.Coerce (coerce)@@ -49,7 +52,8 @@ import Data.String.Here import qualified Data.Text as T import Network.HTTP.Types (status401, status404)-import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.API import Servant.Auth () import Servant.Auth.Client@@ -62,10 +66,15 @@   = PushToken Token   | PushSigningKey Token SigningKey -data PushCache-  = PushCache-      { pushCacheName :: Text,-        pushCacheSecret :: PushSecret+data PushParams m r+  = PushParams+      { pushParamsName :: Text,+        pushParamsSecret :: PushSecret,+        -- | how to report results, (some) errors, and do some things+        pushParamsStrategy :: Text -> PushStrategy m r,+        -- | cachix base url, connection manager, see 'Cachix.Client.URI.defaultCachixBaseUrl', 'Servant.Client.mkClientEnv'+        pushParamsClientEnv :: ClientEnv,+        pushParamsStore :: Store       }  data PushStrategy m r@@ -88,57 +97,49 @@  pushSingleStorePath ::   (MonadMask m, MonadIO m) =>-  -- | cachix base url, connection manager, see 'Cachix.Client.URI.defaultCachixBaseUrl', 'Servant.Client.mkClientEnv'-  ClientEnv ->-  Store ->   -- | details for pushing to cache-  PushCache ->-  -- | how to report results, (some) errors, and do some things-  PushStrategy m r ->+  PushParams m r ->   -- | store path   Text ->   -- | r is determined by the 'PushStrategy'   m r-pushSingleStorePath clientEnv _store cache cb storePath = retryAll $ \retrystatus -> do+pushSingleStorePath cache storePath = retryAll $ \retrystatus -> do   let storeHash = fst $ splitStorePath $ toS storePath-      name = pushCacheName cache+      name = pushParamsName cache+      strategy = pushParamsStrategy cache storePath   -- Check if narinfo already exists   res <--    liftIO $ (`runClientM` clientEnv) $+    liftIO $ (`runClientM` pushParamsClientEnv cache) $       API.narinfoHead         cachixClient-        (getCacheAuthToken cache)+        (getCacheAuthToken (pushParamsSecret cache))         name         (NarInfoHash.NarInfoHash storeHash)   case res of-    Right NoContent -> onAlreadyPresent cb -- we're done as store path is already in the cache+    Right NoContent -> onAlreadyPresent strategy -- we're done as store path is already in the cache     Left err-      | isErr err status404 -> uploadStorePath clientEnv _store cache cb storePath retrystatus-      | isErr err status401 -> on401 cb-      | otherwise -> onError cb err+      | isErr err status404 -> uploadStorePath cache storePath retrystatus+      | isErr err status401 -> on401 strategy+      | otherwise -> onError strategy err -getCacheAuthToken :: PushCache -> Token-getCacheAuthToken cache = case pushCacheSecret cache of-  PushToken token -> token-  PushSigningKey token _ -> token+getCacheAuthToken :: PushSecret -> Token+getCacheAuthToken (PushToken token) = token+getCacheAuthToken (PushSigningKey token _) = token  uploadStorePath ::   (MonadMask m, MonadIO m) =>-  -- | cachix base url, connection manager, see 'Cachix.Client.URI.defaultCachixBaseUrl', 'Servant.Client.mkClientEnv'-  ClientEnv ->-  Store ->   -- | details for pushing to cache-  PushCache ->-  -- | how to report results, (some) errors, and do some things-  PushStrategy m r ->-  -- | store path+  PushParams m r ->   Text ->   RetryStatus ->   -- | r is determined by the 'PushStrategy'   m r-uploadStorePath clientEnv store cache cb storePath retrystatus = do+uploadStorePath cache storePath retrystatus = do   let (storeHash, storeSuffix) = splitStorePath $ toS storePath-      name = pushCacheName cache+      name = pushParamsName cache+      store = pushParamsStore cache+      clientEnv = pushParamsClientEnv cache+      strategy = pushParamsStrategy cache storePath   narSizeRef <- liftIO $ newIORef 0   fileSizeRef <- liftIO $ newIORef 0   narHashRef <- liftIO $ newIORef ("" :: ByteString)@@ -149,9 +150,11 @@   let cmd = proc "nix-store" ["--dump", toS storePath]       storePathSize :: Int64       storePathSize = Store.validPathInfoNarSize pathinfo-  onAttempt cb retrystatus storePathSize-  (ClosedStream, stdoutStream, Inherited, cph) <- liftIO $ streamingProcess cmd-  withXzipCompressor cb $ \xzCompressor -> do+  onAttempt strategy retrystatus storePathSize+  -- create_group makes subprocess ignore signals such as ctrl-c that we handle in haskell main thread+  -- see https://github.com/haskell/process/issues/198+  (ClosedStream, stdoutStream, Inherited, cph) <- liftIO $ streamingProcess (cmd {create_group = True})+  withXzipCompressor strategy $ \xzCompressor -> do     let stream' =           stdoutStream             .| passthroughSizeSink narSizeRef@@ -171,7 +174,7 @@     (_ :: NoContent) <-       liftIO         $ (`withClientM` newClientEnv)-          (API.createNar cachixClient (getCacheAuthToken cache) name (mapOutput coerce stream'))+          (API.createNar cachixClient (getCacheAuthToken (pushParamsSecret cache)) name (mapOutput coerce stream'))         $ escalate           >=> \NoContent -> do             exitcode <- waitForStreamingProcess cph@@ -185,13 +188,13 @@       fileHash <- readIORef fileHashRef       fileSize <- readIORef fileSizeRef       deriver <--        if omitDeriver cb+        if omitDeriver strategy           then pure Store.unknownDeriver           else toS <$> Store.validPathInfoDeriver pathinfo       referencesPathSet <- Store.validPathInfoReferences pathinfo       references <- sort <$> Store.traversePathSet (pure . toS) referencesPathSet       let fp = fingerprint storePath narHash narSize references-          (sig, authToken) = case pushCacheSecret cache of+          (sig, authToken) = case pushParamsSecret cache of             PushToken token -> (Nothing, token)             PushSigningKey token signKey -> (Just $ toS $ B64.encode $ unSignature $ dsign (signingSecretKey signKey) fp, token)           nic =@@ -218,15 +221,7 @@           name           (NarInfoHash.NarInfoHash storeHash)           nic-    onDone cb---- Catches all exceptions except skipAsyncExceptions-retryAll :: (MonadIO m, MonadMask m) => (RetryStatus -> m a) -> m a-retryAll = recoverAll defaultRetryPolicy-  where-    defaultRetryPolicy :: RetryPolicy-    defaultRetryPolicy =-      exponentialBackoff 100000 <> limitRetries 3+    onDone strategy  -- | Push an entire closure --@@ -237,16 +232,19 @@   --   -- For example: @'mapConcurrentlyBounded' 4@   (forall a b. (a -> m b) -> [a] -> m [b]) ->-  -- | See 'pushSingleStorePath'-  ClientEnv ->-  Store ->-  PushCache ->-  (Text -> PushStrategy m r) ->+  PushParams m r ->   -- | Initial store paths   [Text] ->   -- | Every @r@ per store path of the entire closure of store paths   m [r]-pushClosure traversal clientEnv store pushCache pushStrategy inputStorePaths = do+pushClosure traversal pushParams inputStorePaths = do+  missingPaths <- getMissingPathsForClosure pushParams inputStorePaths+  traversal (\path -> retryAll $ \retrystatus -> uploadStorePath pushParams path retrystatus) missingPaths++getMissingPathsForClosure :: (MonadIO m, MonadMask m) => PushParams m r -> [Text] -> m [Text]+getMissingPathsForClosure pushParams inputStorePaths = do+  let store = pushParamsStore pushParams+      clientEnv = pushParamsClientEnv pushParams   -- Get the transitive closure of dependencies   paths <-     liftIO $ do@@ -264,13 +262,11 @@           ( (`runClientM` clientEnv) $               API.narinfoBulk                 cachixClient-                (getCacheAuthToken pushCache)-                (pushCacheName pushCache)+                (getCacheAuthToken (pushParamsSecret pushParams))+                (pushParamsName pushParams)                 (fst . splitStorePath <$> paths)           )-  let missingHashes = Set.fromList missingHashesList-      missingPaths = filter (\path -> Set.member (fst (splitStorePath path)) missingHashes) paths-  traversal (\path -> retryAll $ \retrystatus -> uploadStorePath clientEnv store pushCache (pushStrategy path) path retrystatus) missingPaths+  return $ filter (\path -> Set.member (fst (splitStorePath path)) (Set.fromList missingHashesList)) paths  -- TODO: move to a separate module specific to cli
+ src/Cachix/Client/PushQueue.hs view
@@ -0,0 +1,120 @@+{- Implements a queue with the following properties:++- waits for queue to be fully pushed when exiting using ctrl-c (SIGINT)+- allows stopping the producer+- avoid duplicate pushing of the same store paths++To safetly exit on demand, signal SIGINT.+-}+module Cachix.Client.PushQueue+  ( startWorkers,+    Queue,+  )+where++import qualified Cachix.Client.Push as Push+import Cachix.Client.Retry (retryAll)+import Control.Concurrent.Async+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 Protolude+import qualified System.Posix.Signals as Signals++type StorePath = Text++type Queue = TBQueue.TBQueue StorePath++data PushWorkerState+  = PushWorkerState+      { pushQueue :: Queue,+        inProgress :: TVar Int+      }++data QueryWorkerState+  = QueryWorkerState+      { queryQueue :: Queue,+        alreadyQueued :: S.Set StorePath,+        lock :: Lock.Lock+      }++worker :: Push.PushParams IO () -> PushWorkerState -> IO ()+worker pushParams workerState = forever $ do+  storePath <- atomically $ TBQueue.readTBQueue $ pushQueue workerState+  bracket_ (inProgresModify (+ 1)) (inProgresModify (\x -> x - 1))+    $ retryAll+    $ Push.uploadStorePath pushParams storePath+  where+    inProgresModify f =+      atomically $ modifyTVar' (inProgress workerState) f++-- NOTE: producer is responsible for signaling SIGINT upon termination+-- NOTE: producer should return an `IO ()` that should be a blocking operation for terminating it+startWorkers :: Int -> (Queue -> IO (IO ())) -> Push.PushParams IO () -> IO ()+startWorkers numWorkers mkProducer pushParams = do+  -- start query worker+  (newQueryQueue, newPushQueue, newLock) <-+    atomically $+      (,,) <$> TBQueue.newTBQueue 10000 <*> TBQueue.newTBQueue 10000 <*> Lock.new+  let queryWorkerState = QueryWorkerState newQueryQueue S.empty newLock+  queryWorker <- async $ queryLoop queryWorkerState newPushQueue pushParams+  -- start push workers+  stopProducerCallback <- mkProducer newQueryQueue+  progress <- newTVarIO 0+  let pushWorkerState = PushWorkerState newPushQueue progress+  pushWorker <- async $ replicateConcurrently_ numWorkers $ worker pushParams pushWorkerState+  void $ Signals.installHandler Signals.sigINT (Signals.CatchOnce (exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState)) Nothing+  (_, eitherException) <- waitAnyCatchCancel [pushWorker, queryWorker]+  case eitherException of+    Left exc | fromException exc == Just StopWorker -> return ()+    Left exc -> throwIO exc+    Right () -> return ()++data StopWorker = StopWorker+  deriving (Eq, Show)++instance Exception StopWorker++queryLoop :: QueryWorkerState -> Queue -> Push.PushParams IO () -> IO ()+queryLoop workerState pushqueue pushParams = do+  -- this blocks until item is available and doesn't remove it from the queue+  _ <- atomically $ TBQueue.peekTBQueue (queryQueue workerState)+  (missingStorePathsSet, alreadyQueuedSet) <- Lock.with (lock workerState) $ do+    storePaths <- atomically $ TBQueue.flushTBQueue (queryQueue workerState)+    -- if push queue is empty we can our store path cache here as getClosure will do its job+    alreadyQueuedSet <- atomically $ do+      isEmpty <- TBQueue.isEmptyTBQueue pushqueue+      if isEmpty+        then return S.empty+        else return $ alreadyQueued workerState+    missingStorePaths <- Push.getMissingPathsForClosure pushParams storePaths+    let missingStorePathsSet = S.fromList missingStorePaths+        uncachedMissingStorePaths = S.difference missingStorePathsSet alreadyQueuedSet+    atomically $ for_ uncachedMissingStorePaths $ TBQueue.writeTBQueue pushqueue+    return (missingStorePathsSet, alreadyQueuedSet)+  queryLoop (workerState {alreadyQueued = S.union missingStorePathsSet alreadyQueuedSet}) pushqueue pushParams++exitOnceQueueIsEmpty :: IO () -> Async () -> Async () -> QueryWorkerState -> PushWorkerState -> IO ()+exitOnceQueueIsEmpty stopProducerCallback pushWorker queryWorker queryWorkerState pushWorkerState = do+  putText "Stopped watching /nix/store and waiting for queue to empty ..."+  stopProducerCallback+  go+  where+    go = do+      (isDone, inprogress, queueLength) <- atomically $ do+        pushQueueLength <- TBQueue.lengthTBQueue $ pushQueue pushWorkerState+        queryQueueLength <- TBQueue.lengthTBQueue $ queryQueue queryWorkerState+        inprogress <- readTVar $ inProgress pushWorkerState+        isLocked <- Lock.locked (lock queryWorkerState)+        let isDone = pushQueueLength == 0 && queryQueueLength == 0 && inprogress == 0 && not isLocked+        return (isDone, inprogress, pushQueueLength)+      if isDone+        then do+          putText "Done."+          cancelWith queryWorker StopWorker+          cancelWith pushWorker StopWorker+        else do+          putText $ "Waiting to finish: " <> show inprogress <> " pushing, " <> show queueLength <> " in queue"+          threadDelay (1000 * 1000)+          go
+ src/Cachix/Client/Retry.hs view
@@ -0,0 +1,16 @@+module Cachix.Client.Retry+  ( retryAll,+  )+where++import Control.Exception.Safe (MonadMask)+import Control.Retry (RetryPolicy, RetryStatus, exponentialBackoff, limitRetries, recoverAll)+import Protolude++-- Catches all exceptions except skipAsyncExceptions+retryAll :: (MonadIO m, MonadMask m) => (RetryStatus -> m a) -> m a+retryAll = recoverAll defaultRetryPolicy+  where+    defaultRetryPolicy :: RetryPolicy+    defaultRetryPolicy =+      exponentialBackoff (1000 * 1000) <> limitRetries 5
src/Cachix/Client/Secrets.hs view
@@ -13,7 +13,8 @@ import qualified Data.ByteString.Base64 as B64 import qualified Data.ByteString.Char8 as BC import Data.Char (isSpace)-import Protolude+import Protolude hiding (toS)+import Protolude.Conv  -- | A secret key for signing nars. newtype SigningKey = SigningKey {signingSecretKey :: SecretKey}
src/Cachix/Client/URI.hs view
@@ -8,7 +8,8 @@   ) where -import Protolude+import Protolude hiding (toS)+import Protolude.Conv import Servant.Client import qualified URI.ByteString as UBS import URI.ByteString hiding (Scheme)
+ src/Cachix/Client/WatchStore.hs view
@@ -0,0 +1,34 @@+module Cachix.Client.WatchStore+  ( startWorkers,+  )+where++import Cachix.Client.Push+import qualified Cachix.Client.PushQueue as PushQueue+import qualified Control.Concurrent.STM.TBQueue as TBQueue+import Data.List (isSuffixOf)+import Protolude+import System.FSNotify++startWorkers :: Int -> PushParams IO () -> IO ()+startWorkers numWorkers pushParams = do+  withManager $ \mgr -> PushQueue.startWorkers numWorkers (producer mgr) pushParams++producer :: WatchManager -> PushQueue.Queue -> IO (IO ())+producer mgr queue = do+  putText "Watching /nix/store for new store paths ..."+  watchDir mgr "/nix/store" filterOnlyStorePaths (queueStorePathAction queue)++queueStorePathAction :: PushQueue.Queue -> Event -> IO ()+queueStorePathAction queue (Removed lockFile _ _) = atomically $ TBQueue.writeTBQueue queue (toS $ dropLast 5 lockFile)+queueStorePathAction _ _ = return ()++dropLast :: Int -> [a] -> [a]+dropLast index xs = take (length xs - index) xs++-- we queue store paths after their lock has been removed+filterOnlyStorePaths :: ActionPredicate+filterOnlyStorePaths (Removed fp _ _)+  | ".drv.lock" `isSuffixOf` fp = False+  | ".lock" `isSuffixOf` fp = True+filterOnlyStorePaths _ = False
test/NixConfSpec.hs view
@@ -4,6 +4,7 @@  import Cachix.Client.NixConf as NixConf import Cachix.Types.BinaryCache (BinaryCache (..))+import Cachix.Types.Permission (Permission (..)) import Data.String.Here import Protolude import Test.Hspec@@ -17,6 +18,7 @@     { name = "name",       uri = "https://name.cachix.org",       isPublic = True,+      permission = Admin,       publicSigningKeys = ["pub"],       githubUsername = "foobar"     }