packages feed

cachix 1.6.1 → 1.7

raw patch · 40 files changed

+3263/−640 lines, 40 filesdep +amazonkadep +amazonka-coredep +amazonka-s3dep −cryptonite

Dependencies added: amazonka, amazonka-core, amazonka-s3, attoparsec, crypton, exceptions, generic-lens, inline-c-cpp, microlens, nix-narinfo, transformers, unliftio, unliftio-core

Dependencies removed: cryptonite

Files

CHANGELOG.md view
@@ -5,6 +5,28 @@ 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] - 2023-01-08++## Added++- daemon mode: push to cachix while building++- `cachix import`: allow importing S3 binary caches into Cachix++## Fixed++- Ignore sigPIPE exception++- Fix InvalidPath for case insensitive conflicts on macOS++- Handle common store path errors to print a human readable exception++- Pretty Print C+ exceptions++- Deploy: don't throw errors when bootstrapping fails++- Retry only HTTP exceptions, failing on fatal exceptions (like auth access)+ ## [1.6.1] - 2023-09-25  ## Fixed
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.6.1+version:            1.7 synopsis:   Command-line client for Nix binary cache hosting https://cachix.org @@ -29,13 +29,13 @@ common defaults   build-depends:      base >=4.7 && <5   default-extensions:-    NoImplicitPrelude-    NoImportQualifiedPost     DeriveAnyClass     DeriveGeneric     DerivingVia     LambdaCase     NamedFieldPuns+    NoImplicitPrelude+    NoImportQualifiedPost     OverloadedStrings     RecordWildCards     ScopedTypeVariables@@ -64,12 +64,27 @@     Cachix.Client     Cachix.Client.CNix     Cachix.Client.Commands+    Cachix.Client.Commands.Push     Cachix.Client.Config     Cachix.Client.Config.Orphans     Cachix.Client.Daemon     Cachix.Client.Daemon.Client     Cachix.Client.Daemon.Listen+    Cachix.Client.Daemon.Log+    Cachix.Client.Daemon.PostBuildHook+    Cachix.Client.Daemon.Progress+    Cachix.Client.Daemon.Protocol+    Cachix.Client.Daemon.Push+    Cachix.Client.Daemon.PushManager+    Cachix.Client.Daemon.PushManager.PushJob+    Cachix.Client.Daemon.ShutdownLatch+    Cachix.Client.Daemon.Subscription     Cachix.Client.Daemon.Types+    Cachix.Client.Daemon.Types.Daemon+    Cachix.Client.Daemon.Types.Log+    Cachix.Client.Daemon.Types.PushEvent+    Cachix.Client.Daemon.Types.PushManager+    Cachix.Client.Daemon.Worker     Cachix.Client.Env     Cachix.Client.Exception     Cachix.Client.HumanSize@@ -104,8 +119,12 @@   autogen-modules:   Paths_cachix   build-depends:     , aeson+    , amazonka                >=2.0+    , amazonka-core           >=2.0+    , amazonka-s3             >=2.0     , ascii-progress     , async+    , attoparsec     , base64-bytestring     , bytestring     , cachix-api@@ -115,15 +134,17 @@     , conduit-extra     , conduit-zstd     , containers-    , cryptonite+    , crypton     , deepseq     , dhall                   >=1.28.0     , directory     , ed25519     , either+    , exceptions     , extra     , filepath     , fsnotify                >=0.4.1+    , generic-lens     , hercules-ci-cnix-store     , here     , hnix-store-core         >=0.6.1.0@@ -132,13 +153,16 @@     , http-conduit     , http-types     , immortal+    , inline-c-cpp     , katip     , lukko     , lzma-conduit     , megaparsec              >=7.0.0     , memory+    , microlens     , netrc     , network+    , nix-narinfo     , optparse-applicative     , pretty-terminal     , prettyprinter@@ -159,7 +183,10 @@     , temporary     , text     , time+    , transformers     , unix+    , unliftio+    , unliftio-core     , unordered-containers     , uri-bytestring     , uuid@@ -210,6 +237,9 @@   main-is:            Main.hs   hs-source-dirs:     test   other-modules:+    Daemon.PostBuildHookSpec+    Daemon.PushManagerSpec+    DeploySpec     InstallationModeSpec     NetRcSpec     NixConfSpec@@ -222,14 +252,17 @@     , bytestring     , cachix     , cachix-api+    , containers     , dhall     , directory     , extra     , here     , hspec     , protolude+    , retry     , servant-auth-client     , servant-client-core     , temporary+    , time    build-tool-depends: hspec-discover:hspec-discover
cachix/Main.hs view
@@ -1,8 +1,9 @@ module Main (main) where  import qualified Cachix.Client as CC+import Cachix.Client.CNix (handleCppExceptions) import Cachix.Client.Exception (CachixException)-import Control.Exception.Safe (displayException, handle)+import Control.Exception.Safe (Handler (..), catches, displayException) import GHC.IO.Encoding import System.Exit (exitFailure) import System.IO@@ -15,12 +16,12 @@   hSetBuffering stderr LineBuffering   handleExceptions CC.main -handleExceptions :: IO a -> IO a-handleExceptions = handle handler-  where-    handler :: CachixException -> IO a-    handler e = do-      hPutStrLn stderr ""-      hPutStr stderr (displayException e)-      hFlush stderr-      exitFailure+handleExceptions :: IO () -> IO ()+handleExceptions f = f `catches` [Handler handleCachixExceptions, Handler handleCppExceptions]++handleCachixExceptions :: CachixException -> IO ()+handleCachixExceptions e = do+  hPutStrLn stderr ""+  hPutStr stderr (displayException e)+  hFlush stderr+  exitFailure
src/Cachix/Client.hs view
@@ -6,29 +6,40 @@ import Cachix.Client.Commands as Commands import qualified Cachix.Client.Config as Config import qualified Cachix.Client.Daemon as Daemon+import qualified Cachix.Client.Daemon.Client as Daemon.Client import Cachix.Client.Env (cachixoptions, mkEnv) import Cachix.Client.OptionsParser (CachixCommand (..), DaemonCommand (..), getOpts) import Cachix.Client.Version (cachixVersion) import Cachix.Deploy.ActivateCommand as ActivateCommand import qualified Cachix.Deploy.Agent as AgentCommand import qualified Cachix.Deploy.OptionsParser as DeployOptions+import qualified Hercules.CNix as CNix+import qualified Hercules.CNix.Util as CNix.Util import Protolude import System.Console.AsciiProgress (displayConsoleRegions)+import qualified System.Posix.Signals as Signal  main :: IO () main = displayConsoleRegions $ do   (flags, command) <- getOpts   env <- mkEnv flags++  installSignalHandlers++  initNixStore+   let cachixOptions = cachixoptions env   case command of     AuthToken token -> Commands.authtoken env token     Config configCommand -> Config.run cachixOptions configCommand-    Daemon (DaemonRun daemonOptions pushOptions) -> Daemon.run env daemonOptions pushOptions-    Daemon (DaemonStop daemonOptions) -> Daemon.stop env daemonOptions-    Daemon (DaemonPushPaths daemonOptions cacheName storePaths) -> Daemon.push env daemonOptions cacheName storePaths+    Daemon (DaemonRun daemonOptions pushOptions mcacheName) -> Daemon.start env daemonOptions pushOptions mcacheName+    Daemon (DaemonStop daemonOptions) -> Daemon.Client.stop env daemonOptions+    Daemon (DaemonPushPaths daemonOptions storePaths) -> Daemon.Client.push env daemonOptions storePaths+    Daemon (DaemonWatchExec pushOptions cacheName cmd args) -> Commands.watchExecDaemon env pushOptions cacheName cmd args     GenerateKeypair name -> Commands.generateKeypair env name     Push pushArgs -> Commands.push env pushArgs     Pin pingArgs -> Commands.pin env pingArgs+    Import pushOptions name uri -> Commands.import' env pushOptions name uri     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@@ -36,3 +47,28 @@     Version -> putText cachixVersion     DeployCommand (DeployOptions.Agent opts) -> AgentCommand.run cachixOptions opts     DeployCommand (DeployOptions.Activate opts) -> ActivateCommand.run env opts++-- | Install client-wide signal handlers.+installSignalHandlers :: IO ()+installSignalHandlers = do+  -- Ignore sigPIPE.+  -- By default, sigPIPE will crash the entire program when the reading end of a pipe is closed.+  -- By ignoring sigPIPE, an exception is thrown inline instead, which we can handle.+  _ <- Signal.installHandler Signal.sigPIPE Signal.Ignore Nothing++  return ()++-- | Initialize the Nix library via CNix.+initNixStore :: IO ()+initNixStore = do+  signalset <- Signal.getSignalMask++  -- Initialize the Nix library+  CNix.init++  -- darwin: restore the signal mask modified by Nix+  -- https://github.com/cachix/cachix/issues/501+  Signal.setSignalMask signalset++  -- Interrupt Nix before throwing UserInterrupt+  CNix.Util.installDefaultSigINTHandler
src/Cachix/Client/CNix.hs view
@@ -1,19 +1,97 @@ module Cachix.Client.CNix where -import Hercules.CNix.Store (Store, StorePath, isValidPath, storePathToPath)+import Hercules.CNix.Store (Store, StorePath)+import qualified Hercules.CNix.Store as Store+import Language.C.Inline.Cpp.Exception import Protolude-import System.Console.Pretty (Color (..), color)+import System.Console.Pretty (Color (..), Style (..), color, style) +-- | Checks whether a store path is valid.+validateStorePath :: Store -> StorePath -> IO (Maybe StorePath)+validateStorePath store storePath = do+  isValid <- Store.isValidPath store storePath `catchNixError` const (return False)+  if isValid+    then return (Just storePath)+    else return Nothing++-- | Like 'validateStorePath', but logs a warning when the path is invalid.+filterInvalidStorePath :: Store -> StorePath -> IO (Maybe StorePath)+filterInvalidStorePath store storePath = do+  mstorePath <- validateStorePath store storePath++  when (isNothing mstorePath) (logBadStorePath store storePath)++  return mstorePath+ filterInvalidStorePaths :: Store -> [StorePath] -> IO [Maybe StorePath] filterInvalidStorePaths store =   traverse (filterInvalidStorePath store) -filterInvalidStorePath :: Store -> StorePath -> IO (Maybe StorePath)-filterInvalidStorePath store storePath = do-  isValid <- isValidPath store storePath-  if isValid-    then return $ Just storePath-    else do-      path <- storePathToPath store storePath-      putErrText $ color Yellow $ "Warning: " <> decodeUtf8With lenientDecode path <> " is not valid, skipping"+-- | Follows all symlinks to a store path, returning the final store path.+--+-- Returns Nothing if the path is invalid.+followLinksToStorePath :: Store -> ByteString -> IO (Maybe StorePath)+followLinksToStorePath store storePath =+  (Just <$> Store.followLinksToStorePath store storePath)+    `catchNixError` \e -> do+      logNixError storePath e       return Nothing++data NixError = NixError+  { typ :: Maybe Text,+    msg :: Text+  }+  deriving (Show)++-- | Capture and handle Nix errors+catchNixError :: IO a -> (NixError -> IO a) -> IO a+catchNixError f onError =+  handleJust selectNixError onError f+  where+    selectNixError (CppStdException _eptr msg typ) =+      Just $+        NixError+          { typ = decodeUtf8With lenientDecode <$> typ,+            msg = decodeUtf8With lenientDecode msg+          }+    selectNixError _ = Nothing++-- | Pretty-print a Nix error.+--+-- There might not be a valid store path at this point.+logNixError :: ByteString -> NixError -> IO ()+logNixError storePath (NixError {..}) =+  case typ of+    Just "nix::BadStorePath" -> logBadPath storePath+    _ -> putErrText $ color Red $ style Bold $ "Nix " <> msg++-- | Print a warning when a store path is invalid.+logBadStorePath :: Store -> StorePath -> IO ()+logBadStorePath store storePath = do+  path <- Store.storePathToPath store storePath+  logBadPath path++-- | Print a warning when the path is invalid.+--+-- There are two use-cases when this should be used:+-- 1. Converting a file path to a store path.+-- 2. Filtering out a store path that is not valid.+logBadPath :: ByteString -> IO ()+logBadPath path =+  putErrText $ color Yellow $ "Warning: " <> decodeUtf8With lenientDecode path <> " is not valid, skipping"++-- | Pretty-print unhandled C++ exceptions from Nix.+handleCppExceptions :: CppException -> IO ()+handleCppExceptions e = do+  putErrText ""++  case e of+    CppStdException _eptr msg _t ->+      putErrText $ color Red $ style Bold $ "Nix " <> decodeUtf8With lenientDecode msg+    CppNonStdException _eptr mmsg -> do+      let msg = fromMaybe "unknown exception" mmsg+      putErrText $ color Red $ style Bold $ "C++ exception: " <> decodeUtf8With lenientDecode msg+    CppHaskellException he ->+      putErrText $ color Red $ style Bold $ "Haskell exception: " <> toS (displayException he)++  exitFailure
src/Cachix/Client/Commands.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE DataKinds #-}+{-# LANGUAGE OverloadedLabels #-} {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE TypeOperators #-} @@ -8,64 +9,94 @@     push,     watchStore,     watchExec,+    watchExecDaemon,+    watchExecStore,     use,+    import',     remove,     pin,-    withPushParams,   ) where +import qualified Amazonka+import Amazonka.Data.Body (ResponseBody (..))+import qualified Amazonka.Data.Text+import qualified Amazonka.S3+import Amazonka.S3.GetObject (getObjectResponse_body)+import Amazonka.S3.ListObjectsV2 (listObjectsV2Response_contents)+import Amazonka.S3.Types.Object (object_key) import qualified Cachix.API as API import Cachix.API.Error-import Cachix.Client.CNix (filterInvalidStorePath)+import Cachix.Client.CNix (filterInvalidStorePath, followLinksToStorePath)+import Cachix.Client.Commands.Push import qualified Cachix.Client.Config as Config+import qualified Cachix.Client.Daemon as Daemon+import qualified Cachix.Client.Daemon.PostBuildHook as Daemon.PostBuildHook+import qualified Cachix.Client.Daemon.Progress as Daemon.Progress+import Cachix.Client.Daemon.Types import Cachix.Client.Env (Env (..)) import Cachix.Client.Exception (CachixException (..))-import Cachix.Client.HumanSize (humanSize) import qualified Cachix.Client.InstallationMode as InstallationMode import qualified Cachix.Client.NixConf as NixConf import Cachix.Client.NixVersion (assertNixVersion) import Cachix.Client.OptionsParser-  ( PinOptions (..),+  ( DaemonOptions (..),+    PinOptions (..),     PushArguments (..),     PushOptions (..),   ) import Cachix.Client.Push-import Cachix.Client.Retry (retryAll)+import Cachix.Client.Retry (retryHttp) import Cachix.Client.Secrets   ( SigningKey (SigningKey),     exportSigningKey,   ) import Cachix.Client.Servant+import Cachix.Client.URI (URI)+import qualified Cachix.Client.URI as URI import qualified Cachix.Client.WatchStore as WatchStore-import qualified Cachix.Types.BinaryCache as BinaryCache+import Cachix.Types.BinaryCache (BinaryCacheName) import qualified Cachix.Types.PinCreate as PinCreate import qualified Cachix.Types.SigningKeyCreate as SigningKeyCreate+import Conduit import qualified Control.Concurrent.Async as Async-import Control.Exception.Safe (throwM)-import Control.Retry (RetryStatus (rsIterNumber))+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import Control.Retry (defaultRetryStatus) import Crypto.Sign.Ed25519 (PublicKey (PublicKey), createKeypair)-import qualified Data.ByteString as BS+import qualified Data.Attoparsec.Text import qualified Data.ByteString.Base64 as B64-import qualified Data.Conduit as Conduit+import qualified Data.Conduit.Combinators as C+import Data.Conduit.ConcurrentMap (concurrentMapM_)+import qualified Data.Conduit.List as CL+import qualified Data.Conduit.TMChan as C+import Data.Generics.Labels ()+import Data.HashMap.Strict as HashMap+import Data.IORef import Data.String.Here import qualified Data.Text as T+import Data.Text.IO (hGetLine) import qualified Data.Text.IO as T.IO import GHC.IO.Handle (hDuplicate, hDuplicateTo)-import Hercules.CNix.Store (Store, StorePath, followLinksToStorePath, storePathToPath, withStore)-import Network.HTTP.Types (status401, status404)+import Hercules.CNix.Store (parseStorePath, storePathToPath, withStore)+import Lens.Micro+import qualified Network.HTTP.Client as HTTP+import Network.HTTP.Types (status404)+import qualified Nix.NarInfo as NarInfo import Protolude hiding (toS) import Protolude.Conv-import Servant.API (NoContent)+import Servant.API (NoContent (..)) import Servant.Auth.Client import Servant.Client.Streaming import Servant.Conduit ()-import System.Console.AsciiProgress import System.Console.Pretty import System.Directory (doesFileExist)+import System.Environment (getEnvironment) import System.IO (hIsTerminalDevice)+import System.IO.Error (isEOFError)+import System.IO.Temp (withSystemTempFile) import qualified System.Posix.Signals as Signals import qualified System.Process+import qualified URI.ByteString as UBS  -- TODO: check that token actually authenticates! authtoken :: Env -> Maybe Text -> IO ()@@ -84,7 +115,7 @@       bcc = Config.BinaryCacheConfig name signingKey   -- we first validate if key can be added to the binary cache   (_ :: NoContent) <--    escalate <=< retryAll $ \_ ->+    escalate <=< retryHttp $       (`runClientM` clientenv env) $         API.createKey cachixClient authToken name signingKeyCreate   -- if key was successfully added, write it to the config@@ -112,18 +143,6 @@         Text     ) -notAuthenticatedBinaryCache :: Text -> CachixException-notAuthenticatedBinaryCache name =-  AccessDeniedBinaryCache $-    "Binary cache " <> name <> " doesn't exist or it's private and you need a token: " <> Config.noAuthTokenError--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- getNixEnv :: IO InstallationMode.NixEnv getNixEnv = do   user <- InstallationMode.getUser@@ -142,7 +161,7 @@   optionalAuthToken <- Config.getAuthTokenMaybe (config env)   let token = fromMaybe (Token "") optionalAuthToken   -- 1. get cache public key-  res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient token name+  res <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token name   case res of     Left err -> handleCacheResponse name optionalAuthToken err     Right binaryCache -> do@@ -157,17 +176,6 @@   InstallationMode.removeBinaryCache (Config.hostname $ config env) name $     InstallationMode.getInstallationMode nixEnv InstallationMode.defaultUseOptions -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@@ -182,12 +190,11 @@       -- This is somewhat like the behavior of `cat` for example.       (_, paths) -> return paths   withPushParams env opts name $ \pushParams -> do-    normalized <--      liftIO $-        for inputStorePaths $-          \path -> do-            storePath <- followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 path)-            filterInvalidStorePath (pushParamsStore pushParams) storePath+    normalized <- liftIO $+      for inputStorePaths $ \path ->+        runMaybeT $ do+          storePath <- MaybeT $ followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 path)+          MaybeT $ filterInvalidStorePath (pushParamsStore pushParams) storePath     pushedPaths <-       pushClosure         (mapConcurrentlyBounded (numJobs opts))@@ -201,12 +208,127 @@   throwIO $     DeprecatedCommand "DEPRECATED: cachix watch-store has replaced cachix push --watch-store." +discoverAwsEnv :: Maybe ByteString -> Maybe ByteString -> IO Amazonka.Env+discoverAwsEnv maybeEndpoint maybeRegion = do+  s3Endpoint <-+    case maybeEndpoint of+      Nothing -> pure Amazonka.S3.defaultService+      Just url -> do+        req <- HTTP.parseRequest (toS url)+        pure $+          Amazonka.S3.defaultService+            & Amazonka.setEndpoint (HTTP.secure req) (HTTP.host req) (HTTP.port req)+            -- Don't overwrite requests into the virtual-hosted style i.e. <bucket>.<host>.+            -- This would break IP-based endpoints, like our test endpoint.+            & #s3AddressingStyle .~ Amazonka.S3AddressingStylePath++  region <-+    traverse (escalateAs (FatalError . toS) . Amazonka.Data.Text.fromText . toS) maybeRegion++  -- Create a service client with the custom endpoint+  Amazonka.newEnv Amazonka.discover+    <&> Amazonka.configureService s3Endpoint+      . maybe identity (#region .~) region++import' :: Env -> PushOptions -> Text -> URI -> IO ()+import' env pushOptions name s3uri = do+  awsEnv <- discoverAwsEnv (URI.getQueryParam s3uri "endpoint") (URI.getQueryParam s3uri "region")+  putErrText $ "Importing narinfos/nars using " <> show (numJobs pushOptions) <> " workers from " <> URI.serialize s3uri <> " to " <> name+  putErrText ""+  Amazonka.runResourceT $+    runConduit $+      Amazonka.paginate awsEnv (Amazonka.S3.newListObjectsV2 bucketName)+        .| CL.mapMaybe (^. listObjectsV2Response_contents)+        .| CL.concat+        .| CL.map (^. object_key)+        .| CL.filter (T.isSuffixOf ".narinfo" . Amazonka.Data.Text.toText)+        .| concurrentMapM_ (numJobs pushOptions) (numJobs pushOptions * 2) (uploadNarinfo awsEnv)+        .| CL.sinkNull+  putErrText "All done."+  where+    bucketName :: Amazonka.S3.BucketName+    bucketName = Amazonka.S3.BucketName bucketNameText+    bucketNameText = toS $ UBS.hostBS $ URI.getHostname s3uri++    getObject ::+      (MonadResource m) =>+      Amazonka.Env ->+      Amazonka.S3.ObjectKey ->+      m (ConduitT () ByteString (ResourceT IO) ())+    getObject awsEnv key = do+      rs <- Amazonka.send awsEnv (Amazonka.S3.newGetObject bucketName key)+      return $ body $ rs ^. getObjectResponse_body++    fileHashParse :: Text -> IO Text+    fileHashParse s+      | "sha256:" `T.isPrefixOf` s = return $ T.drop 7 s+      | otherwise = throwM $ ImportUnsupportedHash $ "file hash " <> s <> " is unsupported. Leave us feedback at https://github.com/cachix/cachix/issues/601"++    uploadNarinfo :: Amazonka.Env -> Amazonka.S3.ObjectKey -> ResourceT IO ()+    uploadNarinfo awsEnv entry = liftIO $ do+      let storeHash = T.dropEnd 8 $ Amazonka.Data.Text.toText entry++      -- get narinfo+      narinfoText <- runConduitRes $ do+        narinfoStream <- getObject awsEnv entry+        narinfoStream .| C.decodeUtf8 .| C.fold++      -- parse narinfo+      case Data.Attoparsec.Text.parseOnly NarInfo.parseNarInfo (toS narinfoText) of+        Left e -> hPutStr stderr $ "error while parsing " <> storeHash <> ": " <> show e+        Right parsedNarInfo -> do+          -- we support only sha256: for now+          narInfo <- do+            fileHash <- fileHashParse $ NarInfo.fileHash parsedNarInfo+            return $ parsedNarInfo {NarInfo.fileHash = fileHash}++          -- stream nar and narinfo+          liftIO $ withPushParams env pushOptions name $ \pushParams -> do+            narinfoResponse <- liftIO $ narinfoExists pushParams (toS storeHash)+            let storePathText = NarInfo.storePath narInfo+                store = pushParamsStore pushParams+            storePath <- liftIO $ parseStorePath store (toS storePathText)+            let strategy = pushParamsStrategy pushParams storePath++            case narinfoResponse of+              Right NoContent -> onAlreadyPresent strategy+              Left err+                | isErr err status404 -> runResourceT $ do+                    pathInfo <- newPathInfoFromNarInfo narInfo+                    let fileSize = fromInteger $ NarInfo.fileSize narInfo++                    case readMaybe (NarInfo.compression narInfo) of+                      Nothing -> putErrText $ "Unsupported compression method: " <> NarInfo.compression narInfo+                      Just compressionMethod -> do+                        liftIO $ onAttempt strategy defaultRetryStatus fileSize++                        narStream <- getObject awsEnv $ Amazonka.S3.ObjectKey $ NarInfo.url narInfo++                        res <-+                          runConduit $+                            narStream+                              .| streamCopy pushParams storePath fileSize defaultRetryStatus compressionMethod++                        case res of+                          Left uploadErr ->+                            liftIO $ onError strategy uploadErr+                          Right (uploadResult, uploadNarDetails) -> do+                            -- TODO: Check that the file size matches?+                            -- Copy over details about the NAR from the narinfo.+                            let newNarDetails = uploadNarDetails {undNarSize = NarInfo.narSize narInfo, undNarHash = NarInfo.narHash narInfo}++                            nic <- newNarInfoCreate pushParams storePath pathInfo newNarDetails+                            completeNarUpload pushParams uploadResult nic++                            liftIO $ onDone strategy+                | otherwise -> putErrText $ show err+ pin :: Env -> PinOptions -> IO () pin env pinOpts = do   authToken <- Config.getAuthTokenRequired (config env)   storePath <- withStore $ \store -> do-    path <- followLinksToStorePath store (encodeUtf8 $ pinStorePath pinOpts)-    storePathToPath store path+    mpath <- followLinksToStorePath store (encodeUtf8 $ pinStorePath pinOpts)+    maybe exitFailure (storePathToPath store) mpath   traverse_ (validateArtifact (toS storePath)) (pinArtifacts pinOpts)   let pinCreate =         PinCreate.PinCreate@@ -215,8 +337,10 @@             artifacts = pinArtifacts pinOpts,             keep = pinKeep pinOpts           }-  void $ escalate <=< retryAll $ \_ ->-    (`runClientM` clientenv env) $ API.createPin cachixClient authToken (pinCacheName pinOpts) pinCreate+  void $+    escalate <=< retryHttp $+      (`runClientM` clientenv env) $+        API.createPin cachixClient authToken (pinCacheName pinOpts) pinCreate   where     validateArtifact :: Text -> Text -> IO ()     validateArtifact storePath artifact = do@@ -230,123 +354,138 @@   withPushParams env opts name $ \pushParams ->     WatchStore.startWorkers (pushParamsStore pushParams) (numJobs opts) pushParams -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-          }-      watch = do-        hDuplicateTo stderr stdout -- redirect all stdout to stderr-        WatchStore.startWorkers (pushParamsStore pushParams) (numJobs pushOpts) pushParams+-- | Run a command and upload any new paths to the binary cache.+--+-- Registers a post-build hook if the user is trusted.+-- Otherwise, falls back to watching the entire Nix store.+watchExec :: Env -> PushOptions -> BinaryCacheName -> Text -> [Text] -> IO ()+watchExec env pushOptions cacheName cmd args = do+  nixEnv <- getNixEnv -  (_, exitCode) <--    Async.concurrently watch $ do-      exitCode <--        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-              Signals.raiseSignal Signals.sigINT-          )-          System.Process.waitForProcess+  if InstallationMode.isTrusted nixEnv+    then watchExecDaemon env pushOptions cacheName cmd args+    else do+      putErrText fallbackWarning+      watchExecStore env pushOptions cacheName cmd args+  where+    fallbackWarning =+      color Yellow "WARNING: " <> "failed to register a post-build hook for this command because you're not a trusted user. Falling back to watching the entire Nix store for new paths." -      -- Stop watching the store and wait for all paths to be pushed-      Signals.raiseSignal Signals.sigINT-      return exitCode+-- | Run a command and push any new paths to the binary cache.+--+-- Requires the user to be a trusted user in a multi-user installation.+watchExecDaemon :: Env -> PushOptions -> BinaryCacheName -> Text -> [Text] -> IO ()+watchExecDaemon env pushOpts cacheName cmd args =+  Daemon.PostBuildHook.withSetup Nothing $ \daemonSock nixConfEnv ->+    withSystemTempFile "daemon-log-capture" $ \_ logHandle -> do+      let daemonOptions = DaemonOptions {daemonSocketPath = Just daemonSock}+      daemon <- Daemon.new env daemonOptions (Just logHandle) pushOpts cacheName -  exitWith exitCode-  where-    getProcessHandle (_, _, _, processHandle) = processHandle+      -- Launch the daemon in the background+      daemonThread <- Async.async $ Daemon.run daemon -pushStrategy :: Store -> Maybe Token -> PushOptions -> Text -> BinaryCache.CompressionMethod -> StorePath -> PushStrategy IO ()-pushStrategy store authToken opts name compressionMethod storePath =-  PushStrategy-    { onAlreadyPresent = pass,-      on401 = handleCacheResponse name authToken,-      onError = throwM,-      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-    }-  where-    retryText :: RetryStatus -> Text-    retryText retryStatus =-      if rsIterNumber retryStatus == 0-        then ""-        else color Yellow $ "retry #" <> show (rsIterNumber retryStatus) <> " "+      -- Subscribe to all push events+      daemonChan <- Daemon.subscribe daemon -    showUploadProgress retryStatus size = do-      let hSize = toS $ humanSize $ fromIntegral size-      path <- liftIO $ decodeUtf8With lenientDecode <$> storePathToPath store storePath+      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 -      isTerminal <- liftIO $ hIsTerminalDevice stdout-      onTick <--        if isTerminal-          then do-            let bar = color Blue "[:bar] " <> toS (retryText retryStatus) <> toS path <> " (:percent of " <> hSize <> ")"-                barLength = T.length $ T.replace ":percent" "  0%" (T.replace "[:bar]" "" (toS bar))+      -- TODO: process and fold events into a state during command execution+      daemonExitCode <- Async.withAsync (postWatchExec daemonChan) $ \_ -> do+        Daemon.stopIO daemon+        Async.wait daemonThread -            progressBar <--              liftIO $-                newProgressBar-                  def-                    { pgTotal = fromIntegral size,-                      -- https://github.com/yamadapc/haskell-ascii-progress/issues/24-                      pgWidth = 20 + barLength,-                      pgOnCompletion = Just $ color Green "✓ " <> toS path <> " (" <> hSize <> ")",-                      pgFormat = bar-                    }+      -- Print the daemon log in case there was an internal error+      case daemonExitCode of+        ExitFailure _ -> printLog logHandle+        ExitSuccess -> return () -            return $ liftIO . tickN progressBar . BS.length-          else do-            -- we append newline instead of putStrLn due to https://github.com/haskell/text/issues/242-            appendErrText $ retryText retryStatus <> "Pushing " <> path <> " (" <> toS hSize <> ")\n"-            return $ const pass+      exitWith exitCode+  where+    postWatchExec chan = do+      statsRef <- newIORef HashMap.empty+      runConduit $+        C.sourceTMChan chan+          .| C.mapM_ (displayPushEvent statsRef) -      Conduit.awaitForever $ \chunk -> do-        Conduit.yield chunk-        onTick chunk+    -- Deduplicate events by path and retry status+    displayPushEvent statsRef PushEvent {eventMessage} = liftIO $ do+      stats <- readIORef statsRef+      case eventMessage of+        PushStorePathAttempt path pathSize retryStatus -> do+          case HashMap.lookup path stats of+            Just (progress, prevRetryStatus)+              | prevRetryStatus == retryStatus -> return ()+              | otherwise -> do+                  newProgress <- Daemon.Progress.update progress retryStatus+                  writeIORef statsRef $ HashMap.insert path (newProgress, retryStatus) stats+            Nothing -> do+              progress <- Daemon.Progress.new stderr (toS path) pathSize retryStatus+              writeIORef statsRef $ HashMap.insert path (progress, retryStatus) stats+        PushStorePathProgress path _ newBytes -> do+          case HashMap.lookup path stats of+            Nothing -> return ()+            Just (progress, _) -> Daemon.Progress.tick progress newBytes+        PushStorePathDone path -> do+          mapM_ (Daemon.Progress.complete . fst) (HashMap.lookup path stats)+        PushStorePathFailed path _ -> do+          mapM_ (Daemon.Progress.fail . fst) (HashMap.lookup path stats)+        _ -> return () -withPushParams :: Env -> PushOptions -> Text -> (PushParams IO () -> IO ()) -> IO ()-withPushParams env pushOpts name m = do-  pushSecret <- findPushSecret (config env) name-  authToken <- Config.getAuthTokenMaybe (config env)-  compressionMethodBackend <- case pushSecret of-    PushSigningKey {} -> pure Nothing-    PushToken {} -> do-      let token = fromMaybe (Token "") authToken-      res <- retryAll $ \_ -> (`runClientM` clientenv env) $ API.getCache cachixClient token name-      case res of-        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 ->-    m-      PushParams-        { pushParamsName = name,-          pushParamsSecret = pushSecret,-          pushParamsClientEnv = clientenv env,-          pushOnClosureAttempt = \full missing -> do-            unless (null missing) $ do-              let numMissing = length missing-                  numCached = length full - numMissing-              putErrText $ "Pushing " <> show numMissing <> " paths (" <> show numCached <> " are already present) using " <> T.toLower (show compressionMethod) <> " to cache " <> name <> " ⏳\n"-            return missing,-          pushParamsStrategy = pushStrategy store authToken pushOpts name compressionMethod,-          pushParamsStore = store-        }+    printLog h = getLineLoop+      where+        getLineLoop = do+          eline <- try $ hGetLine h+          case eline of+            Left e+              | isEOFError e -> return ()+              | otherwise -> putErrText $ "Error reading daemon log: " <> show e+            Right line -> do+              hPutStr stderr line+              getLineLoop --- | Put text to stderr without a new line.+-- | Runs a command while watching the entire Nix store and pushing any new paths. ----- This is safe to use when printing from multiple threads, unlike hPutStrLn which may fail to insert the newline at the right place.-appendErrText :: (MonadIO m) => Text -> m ()-appendErrText = hPutStr stderr+-- Prefer to use the more granular 'watchExecDaemon' whenever possible.+watchExecStore :: Env -> PushOptions -> BinaryCacheName -> Text -> [Text] -> IO ()+watchExecStore 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+            }+        watch = do+          hDuplicateTo stderr stdout -- redirect all stdout to stderr+          WatchStore.startWorkers (pushParamsStore pushParams) (numJobs pushOpts) pushParams++    (_, exitCode) <-+      Async.concurrently watch $ do+        exitCode <-+          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+                Signals.raiseSignal Signals.sigINT+            )+            System.Process.waitForProcess++        -- Stop watching the store and wait for all paths to be pushed+        Signals.raiseSignal Signals.sigINT+        return exitCode++    exitWith exitCode+  where+    getProcessHandle (_, _, _, processHandle) = processHandle
+ src/Cachix/Client/Commands/Push.hs view
@@ -0,0 +1,210 @@+{-# LANGUAGE QuasiQuotes #-}++module Cachix.Client.Commands.Push+  ( pushStrategy,+    withPushParams,+    withPushParams',+    handleCacheResponse,+    getPushSecret,+    getPushSecretRequired,+  )+where++import qualified Cachix.API as API+import qualified Cachix.Client.Config as Config+import Cachix.Client.Env (Env (..))+import Cachix.Client.Exception (CachixException (..))+import Cachix.Client.HumanSize (humanSize)+import Cachix.Client.OptionsParser+  ( PushOptions (..),+  )+import Cachix.Client.Push+import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Secrets+import Cachix.Client.Servant+import Cachix.Types.BinaryCache (BinaryCacheName)+import qualified Cachix.Types.BinaryCache as BinaryCache+import Control.Exception.Safe (throwM)+import Control.Retry (RetryStatus (rsIterNumber))+import qualified Data.ByteString as BS+import qualified Data.Conduit as Conduit+import Data.String.Here+import qualified Data.Text as T+import Hercules.CNix (StorePath)+import Hercules.CNix.Store (Store, storePathToPath, withStore)+import Network.HTTP.Types (status401, status404)+import Protolude hiding (toS)+import Protolude.Conv+import Servant.Auth ()+import Servant.Auth.Client+import Servant.Client.Streaming+import Servant.Conduit ()+import System.Console.AsciiProgress+import System.Console.Pretty+import System.Environment (lookupEnv)+import System.IO (hIsTerminalDevice)++pushStrategy :: Store -> Maybe Token -> PushOptions -> Text -> BinaryCache.CompressionMethod -> StorePath -> PushStrategy IO ()+pushStrategy store authToken opts name compressionMethod storePath =+  PushStrategy+    { onAlreadyPresent = pass,+      on401 = handleCacheResponse name authToken,+      onError = throwM,+      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+    }+  where+    retryText :: RetryStatus -> Text+    retryText retryStatus =+      if rsIterNumber retryStatus == 0+        then ""+        else color Yellow $ "retry #" <> show (rsIterNumber retryStatus) <> " "++    showUploadProgress retryStatus size = do+      let hSize = toS $ humanSize $ fromIntegral size+      path <- liftIO $ decodeUtf8With lenientDecode <$> storePathToPath store storePath++      isTerminal <- liftIO $ hIsTerminalDevice stdout+      onTick <-+        if isTerminal+          then do+            let bar = color Blue "[:bar] " <> toS (retryText retryStatus) <> toS path <> " (:percent of " <> hSize <> ")"+                barLength = T.length $ T.replace ":percent" "  0%" (T.replace "[:bar]" "" (toS bar))++            progressBar <-+              liftIO $+                newProgressBar+                  def+                    { pgTotal = fromIntegral size,+                      -- https://github.com/yamadapc/haskell-ascii-progress/issues/24+                      pgWidth = 20 + barLength,+                      pgOnCompletion = Just $ color Green "✓ " <> toS path <> " (" <> hSize <> ")",+                      pgFormat = bar+                    }++            return $ liftIO . tickN progressBar . BS.length+          else do+            -- we append newline instead of putStrLn due to https://github.com/haskell/text/issues/242+            appendErrText $ retryText retryStatus <> "Pushing " <> path <> " (" <> toS hSize <> ")\n"+            return $ const pass++      Conduit.awaitForever $ \chunk -> do+        Conduit.yield chunk+        onTick chunk++withPushParams :: Env -> PushOptions -> BinaryCacheName -> (PushParams IO () -> IO ()) -> IO ()+withPushParams env pushOpts name m = do+  pushSecret <- getPushSecretRequired (config env) name+  withPushParams' env pushOpts name pushSecret m++withPushParams' :: Env -> PushOptions -> BinaryCacheName -> PushSecret -> (PushParams IO () -> IO ()) -> IO ()+withPushParams' env pushOpts name pushSecret m = do+  let authToken = getAuthTokenFromPushSecret pushSecret++  compressionMethodBackend <- case pushSecret of+    PushSigningKey {} -> pure Nothing+    PushToken token -> do+      res <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token name+      case res of+        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 ->+    m+      PushParams+        { pushParamsName = name,+          pushParamsSecret = pushSecret,+          pushParamsClientEnv = clientenv env,+          pushOnClosureAttempt = \full missing -> do+            unless (null missing) $ do+              let numMissing = length missing+                  numCached = length full - numMissing+              putErrText $ "Pushing " <> show numMissing <> " paths (" <> show numCached <> " are already present) using " <> T.toLower (show compressionMethod) <> " to cache " <> name <> " ⏳\n"+            return missing,+          pushParamsStrategy = pushStrategy store authToken pushOpts name compressionMethod,+          pushParamsStore = store+        }++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++notAuthenticatedBinaryCache :: Text -> CachixException+notAuthenticatedBinaryCache name =+  AccessDeniedBinaryCache $+    "Binary cache " <> name <> " doesn't exist or it's private and you need a token: " <> Config.noAuthTokenError++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++-- | Fetch the push credentials from the environment or config.+getPushSecret ::+  Config.Config ->+  -- | Cache name+  Text ->+  IO (Either Text PushSecret)+getPushSecret config name = do+  maybeAuthToken <- Config.getAuthTokenMaybe config++  maybeSigningKeyEnv <- toS <<$>> lookupEnv "CACHIX_SIGNING_KEY"+  let maybeSigningKeyConfig = Config.secretKey <$> head (getBinaryCache config)++  case maybeSigningKeyEnv <|> maybeSigningKeyConfig of+    Just signingKey ->+      return $ PushSigningKey (fromMaybe (Token "") maybeAuthToken) <$> parseSigningKeyLenient signingKey+    Nothing -> case maybeAuthToken of+      Just authToken -> return $ Right $ PushToken authToken+      Nothing -> return $ Left msg+  where+    -- we reverse list of caches to prioritize keys added as last+    getBinaryCache c =+      reverse $+        filter (\bc -> Config.name bc == name) (Config.binaryCaches c)++    msg :: Text+    msg =+      [iTrim|+Neither auth token nor signing key are present.++They are looked up via $CACHIX_AUTH_TOKEN and $CACHIX_SIGNING_KEY,+and if missing also looked up from ~/.config/cachix/cachix.dhall++Read https://mycache.cachix.org for instructions how to push to your binary cache.+    |]++-- | Like 'getPushSecret', but throws a fatal error if the secret is not found.+getPushSecretRequired ::+  Config.Config ->+  -- | Cache name+  Text ->+  -- | Secret key or exception+  IO PushSecret+getPushSecretRequired config name = do+  epushSecret <- getPushSecret config name+  case epushSecret of+    -- TODO: technically, we're missing any credentials, not just the signing key+    Left err -> throwIO $ NoSigningKey err+    Right pushSecret -> return pushSecret++-- | Put text to stderr without a new line.+--+-- This is safe to use when printing from multiple threads, unlike hPutStrLn which may fail to insert the newline at the right place.+appendErrText :: (MonadIO m) => Text -> m ()+appendErrText = hPutStr stderr
src/Cachix/Client/Config.hs view
@@ -7,6 +7,7 @@     parser,     Command (..),     CachixOptions (..),+    defaultCachixOptions,     -- Auth token helpers     getAuthTokenRequired,     getAuthTokenMaybe,@@ -58,6 +59,16 @@     verbose :: Bool   }   deriving (Show)++defaultCachixOptions :: IO CachixOptions+defaultCachixOptions = do+  configPath <- getDefaultFilename+  return $+    CachixOptions+      { host = URI.defaultCachixURI,+        configPath = configPath,+        verbose = False+      }  data Config = Config   { authToken :: Token,
src/Cachix/Client/Daemon.hs view
@@ -1,105 +1,185 @@-{-# LANGUAGE QuasiQuotes #-}- module Cachix.Client.Daemon-  ( run,-    runWithSocket,-    push,+  ( Types.Daemon,+    Types.runDaemon,+    new,+    start,+    run,     stop,+    stopIO,+    subscribe,   ) where -import Cachix.Client.CNix (filterInvalidStorePath)-import qualified Cachix.Client.Commands as Commands+import qualified Cachix.Client.Commands.Push as Commands.Push+import qualified Cachix.Client.Config as Config import Cachix.Client.Config.Orphans ()-import Cachix.Client.Daemon.Client (push, stop) import Cachix.Client.Daemon.Listen as Daemon-import Cachix.Client.Daemon.Types+import qualified Cachix.Client.Daemon.Log as Log+import Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.Push as Push+import qualified Cachix.Client.Daemon.PushManager as PushManager+import Cachix.Client.Daemon.ShutdownLatch+import Cachix.Client.Daemon.Subscription as Subscription+import Cachix.Client.Daemon.Types as Types+import qualified Cachix.Client.Daemon.Types.PushManager as PushManager+import qualified Cachix.Client.Daemon.Worker as Worker import Cachix.Client.Env as Env-import Cachix.Client.OptionsParser as Options+import Cachix.Client.OptionsParser (DaemonOptions, PushOptions)+import qualified Cachix.Client.OptionsParser as Options import Cachix.Client.Push-import Control.Concurrent.Extra (once)-import Control.Concurrent.STM.TBMQueue-import qualified Control.Immortal as Immortal-import Data.String.Here (i)+import Cachix.Types.BinaryCache (BinaryCacheName)+import qualified Cachix.Types.BinaryCache as BinaryCache+import Control.Concurrent.STM.TMChan+import Control.Exception.Safe (catchAny)+import qualified Control.Monad.Catch as E import qualified Data.Text as T-import qualified Hercules.CNix.Store as Store+import qualified Katip import qualified Network.Socket as Socket import Protolude import System.Posix.Process (getProcessID)+import qualified UnliftIO.Async as Async -run :: Env -> DaemonOptions -> PushOptions -> IO ()-run env daemonOptions pushOptions = do-  socketPath <- maybe getSocketPath pure (daemonSocketPath daemonOptions)-  runWithSocket env pushOptions socketPath-    `finally` putErrText "Daemon shut down."+-- | Configure a new daemon. Use 'run' to start it.+new ::+  -- | The Cachix environment.+  Env ->+  -- | Daemon-specific options.+  DaemonOptions ->+  -- | An optional handle to output logs to.+  Maybe Handle ->+  -- | Push options, like compression settings and number of jobs.+  PushOptions ->+  -- | The name of the binary cache to push to.+  BinaryCacheName ->+  -- | The configured daemon environment.+  IO DaemonEnv+new daemonEnv daemonOptions daemonLogHandle daemonPushOptions daemonCacheName = do+  let daemonLogLevel =+        if Config.verbose (Env.cachixoptions daemonEnv)+          then Debug+          else Info+  daemonLogger <- Log.new "cachix.daemon" daemonLogHandle daemonLogLevel -runWithSocket :: Env -> PushOptions -> FilePath -> IO ()-runWithSocket env pushOptions socketPath = do-  -- Create a queue of push requests for the workers to process-  queue <- newTBMQueueIO 1000+  daemonSocketPath <- maybe getSocketPath pure (Options.daemonSocketPath daemonOptions)+  daemonShutdownLatch <- newShutdownLatch+  daemonPid <- getProcessID -  bracketOnError (startWorker queue) identity $ \shutdownWorker -> do-    -- TODO: retry the connection on socket errors-    bracketOnError (Daemon.openSocket socketPath) Socket.close $ \sock -> do-      Socket.listen sock Socket.maxListenQueue+  daemonPushSecret <- Commands.Push.getPushSecretRequired (config daemonEnv) daemonCacheName+  let authToken = getAuthTokenFromPushSecret daemonPushSecret+  daemonBinaryCache <- Push.getBinaryCache daemonEnv authToken daemonCacheName -      putText =<< readyMessage socketPath-      clientStopConn <- Daemon.listen queue sock+  daemonSubscriptionManager <- Subscription.newSubscriptionManager+  let onPushEvent = Subscription.pushEvent daemonSubscriptionManager+  daemonPushManager <- PushManager.newPushManagerEnv daemonPushOptions daemonLogger onPushEvent -      -- Gracefully shutdown the worker before closing the socket-      -- TODO: consider shutdown from Network.Socket-      shutdownWorker+  return $ DaemonEnv {..} -      Socket.gracefulClose clientStopConn 5000-  where-    startWorker queue = do-      worker <- Immortal.create $ \thread ->-        Immortal.onUnexpectedFinish thread logWorkerException $-          runWorker env pushOptions queue+-- | Configure and run the daemon. Equivalent to running 'new' and 'run'.+start :: Env -> DaemonOptions -> PushOptions -> BinaryCacheName -> IO ()+start daemonEnv daemonOptions daemonPushOptions daemonCacheName = do+  daemon <- new daemonEnv daemonOptions Nothing daemonPushOptions daemonCacheName+  void $ run daemon -      once $ do-        putErrText "Shutting down daemon..."-        atomically $ closeTBMQueue queue-        Immortal.mortalize worker-        putErrText "Waiting for worker to finish..."-        Immortal.wait worker+-- | Run a daemon from a given configuration+run :: DaemonEnv -> IO ExitCode+run daemon = runDaemon daemon $ flip E.onError (return $ ExitFailure 1) $ do+  Katip.logFM Katip.InfoS "Starting Cachix Daemon"+  DaemonEnv {..} <- ask -logWorkerException :: Either SomeException () -> IO ()-logWorkerException (Left err) =-  putErrText $ "Exception in daemon worker thread: " <> show err-logWorkerException _ = return ()+  config <- showConfiguration+  Katip.logFM Katip.InfoS $ Katip.ls $ "Configuration:\n" <> config -runWorker :: Env -> PushOptions -> TBMQueue QueuedPushRequest -> IO ()-runWorker env pushOptions queue = loop-  where-    loop =-      atomically (readTBMQueue queue) >>= \case-        Nothing -> return ()-        Just msg -> do-          handleRequest env pushOptions msg-          loop+  let workerCount = Options.numJobs daemonPushOptions+      startWorkers pushParams =+        Worker.startWorkers+          workerCount+          (PushManager.pmTaskQueue daemonPushManager)+          (liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask pushParams) -handleRequest :: Env -> PushOptions -> QueuedPushRequest -> IO ()-handleRequest env pushOptions (QueuedPushRequest {..}) = do-  Commands.withPushParams env pushOptions (binaryCacheName pushRequest) $ \pushParams -> do-    normalized <--      for (storePaths pushRequest) $ \fp -> do-        storePath <- Store.followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 $ T.pack fp)-        filterInvalidStorePath (pushParamsStore pushParams) storePath+  subscriptionManagerThread <-+    liftIO $ Async.async $ runSubscriptionManager daemonSubscriptionManager -    void $-      pushClosure-        (mapConcurrentlyBounded (numJobs pushOptions))-        pushParams-        (catMaybes normalized)+  let shutdownQueue =+        liftIO $ PushManager.stopPushManager daemonPushManager -readyMessage :: FilePath -> IO Text-readyMessage socketPath = do-  -- Get the PID of the process-  pid <- getProcessID-  return-    [i|-Cachix Daemon is ready to push store paths.-PID: ${show pid :: Text}-Listening on socket: ${socketPath}-  |]+  Push.withPushParams $ \pushParams ->+    E.bracketOnError (startWorkers pushParams) Worker.stopWorkers $ \workers -> do+      flip E.onError shutdownQueue $+        -- TODO: retry the connection on socket errors+        E.bracketOnError (Daemon.openSocket daemonSocketPath) Daemon.closeSocket $ \sock -> do+          liftIO $ Socket.listen sock Socket.maxListenQueue++          res <-+            Async.race (waitForShutdown daemonShutdownLatch) $+              Daemon.listen queueJob sock `E.finally` stop++          waitForShutdown daemonShutdownLatch++          Katip.logFM Katip.InfoS "Shutting down daemon..."++          queuedStorePathCount <- PushManager.runPushManager daemonPushManager 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 ()++          shutdownQueue++          -- Gracefully shutdown the worker *before* closing the socket+          Worker.stopWorkers workers++          liftIO $ stopSubscriptionManager daemonSubscriptionManager+          Async.wait subscriptionManagerThread++          -- TODO: say goodbye to all clients waiting for their push to go through+          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 ()++          return ExitSuccess++stop :: Daemon ()+stop = asks daemonShutdownLatch >>= initiateShutdown++stopIO :: DaemonEnv -> IO ()+stopIO DaemonEnv {daemonShutdownLatch} =+  initiateShutdown daemonShutdownLatch++queueJob :: Protocol.PushRequest -> Socket.Socket -> Daemon ()+queueJob pushRequest _clientConn = do+  DaemonEnv {..} <- ask+  -- TODO: subscribe the socket to updates if requested++  -- Queue the job+  void $+    PushManager.runPushManager daemonPushManager (PushManager.addPushJob pushRequest)++subscribe :: DaemonEnv -> IO (TMChan PushEvent)+subscribe DaemonEnv {..} = do+  chan <- liftIO newBroadcastTMChanIO+  liftIO $ atomically $ do+    subscribeToAllSTM daemonSubscriptionManager (SubChannel chan)+    dupTMChan chan++-- | Print debug information about the daemon configuration+showConfiguration :: Daemon Text+showConfiguration = do+  DaemonEnv {..} <- ask+  pure $+    T.intercalate+      "\n"+      [ "PID: " <> show daemonPid,+        "Socket: " <> toS daemonSocketPath,+        "Workers: " <> show (Options.numJobs daemonPushOptions),+        "Cache name: " <> toS daemonCacheName,+        "Cache URI: " <> BinaryCache.uri daemonBinaryCache,+        "Cache public keys: " <> show (BinaryCache.publicSigningKeys daemonBinaryCache),+        "Cache is public: " <> show (BinaryCache.isPublic daemonBinaryCache),+        "Compression: " <> show (Push.getCompressionMethod daemonPushOptions daemonBinaryCache)+      ]
src/Cachix/Client/Daemon/Client.hs view
@@ -1,13 +1,15 @@ module Cachix.Client.Daemon.Client (push, stop) where -import Cachix.Client.Config as Config import Cachix.Client.Daemon.Listen (getSocketPath)-import Cachix.Client.Daemon.Types (ClientMessage (..), PushRequest (..))+import Cachix.Client.Daemon.Protocol as Protocol import Cachix.Client.Env as Env import Cachix.Client.OptionsParser (DaemonOptions (..))-import Cachix.Types.BinaryCache (BinaryCacheName)+import qualified Control.Concurrent.Async as Async+import Control.Exception.Safe (catchAny) import qualified Data.Aeson as Aeson+import qualified Data.ByteString as BS import qualified Network.Socket as Socket+import qualified Network.Socket.ByteString as Socket.BS import qualified Network.Socket.ByteString.Lazy as Socket.LBS import Protolude import qualified System.Posix.IO as Posix@@ -15,32 +17,49 @@ -- | Queue up push requests with the daemon -- -- TODO: wait for the daemon to respond that it has received the request-push :: Env -> DaemonOptions -> BinaryCacheName -> [FilePath] -> IO ()-push Env {config, cachixoptions} daemonOptions cacheName storePaths = do-  sock <- connectToDaemon (daemonSocketPath daemonOptions)-  Socket.LBS.sendAll sock (Aeson.encode pushRequest)+push :: Env -> DaemonOptions -> [FilePath] -> IO ()+push _env daemonOptions storePaths =+  withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do+    Socket.LBS.sendAll sock (Aeson.encode pushRequest)   where     pushRequest =-      ClientPushRequest $-        PushRequest (Config.authToken config) cacheName (Config.host cachixoptions) storePaths+      Protocol.ClientPushRequest $+        PushRequest {storePaths = storePaths} --- | Tell the daemon to stop and wait for it gracefully exit+-- | Tell the daemon to stop and wait for it to gracefully exit stop :: Env -> DaemonOptions -> IO ()-stop _env daemonOptions = do-  sock <- connectToDaemon (daemonSocketPath daemonOptions)-  Socket.LBS.sendAll sock (Aeson.encode ClientStop)+stop _env daemonOptions =+  withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do+    Async.concurrently_ (waitForResponse sock) $+      Socket.LBS.sendAll sock (Aeson.encode Protocol.ClientStop)+  where+    waitForResponse sock = do+      -- Wait for the socket to close+      bs <- Socket.BS.recv sock 4096 `catchAny` (\_ -> return BS.empty) -  -- Wait for the socket to close-  void $ Socket.LBS.recv sock 1+      -- A zero-length response means that the daemon has closed the socket+      guard $ not $ BS.null bs -connectToDaemon :: Maybe FilePath -> IO Socket.Socket-connectToDaemon optionalSocketPath = do+      case Aeson.eitherDecodeStrict bs of+        Left err -> putErrText (toS err)+        Right DaemonBye -> return ()++withDaemonConn :: Maybe FilePath -> (Socket.Socket -> IO a) -> IO a+withDaemonConn optionalSocketPath f = do   socketPath <- maybe getSocketPath pure optionalSocketPath-  sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol-  Socket.connect sock (Socket.SockAddrUnix socketPath)+  bracket (open socketPath) Socket.close f `onException` failedToConnectTo socketPath+  where+    open socketPath = do+      sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol+      Socket.connect sock (Socket.SockAddrUnix socketPath) -  -- Network.Socket.accept sets the socket to non-blocking by default.-  Socket.withFdSocket sock $ \fd ->-    Posix.setFdOption (fromIntegral fd) Posix.NonBlockingRead False+      -- Network.Socket.accept sets the socket to non-blocking by default.+      Socket.withFdSocket sock $ \fd ->+        Posix.setFdOption (fromIntegral fd) Posix.NonBlockingRead False -  return sock+      return sock++    failedToConnectTo :: FilePath -> IO ()+    failedToConnectTo socketPath = do+      putErrText "Failed to connect to Cachix Daemon"+      putErrText $ "Tried to connect to: " <> toS socketPath <> "\n"
src/Cachix/Client/Daemon/Listen.hs view
@@ -1,17 +1,20 @@ module Cachix.Client.Daemon.Listen   ( listen,+    serverBye,     getSocketPath,     openSocket,+    closeSocket,   ) where  import Cachix.Client.Config.Orphans ()-import Cachix.Client.Daemon.Types-import Control.Concurrent.STM.TBMQueue+import Cachix.Client.Daemon.Protocol as Protocol+import Control.Exception.Safe (catchAny) import qualified Control.Exception.Safe as Safe import qualified Data.Aeson as Aeson import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.BS+import qualified Network.Socket.ByteString.Lazy as Socket.LBS import Protolude import System.Directory   ( XdgDirectory (..),@@ -34,27 +37,27 @@     DecodingError err -> "Failed to decode request: " <> toS err  -- | The main daemon server loop.------ The daemon listens for incoming push requests on the provided socket and queues them up for the worker thread.-listen :: TBMQueue QueuedPushRequest -> Socket.Socket -> IO Socket.Socket-listen queue sock = loop+listen :: (MonadIO m) => (Protocol.PushRequest -> Socket.Socket -> m ()) -> Socket.Socket -> m Socket.Socket+listen pushToQueue sock = loop   where     loop = do-      result <- runExceptT $ readPushRequest sock+      result <- liftIO $ runExceptT $ readPushRequest sock        case result of         Right (ClientStop, clientConn) -> do-          putText "Received stop request, shutting down..."           return clientConn         Right (ClientPushRequest pushRequest, clientConn) -> do-          let queuedRequest = QueuedPushRequest pushRequest clientConn-          atomically $ writeTBMQueue queue queuedRequest+          pushToQueue pushRequest clientConn           loop-        Left (DecodingError err) -> do-          putErrText $ "Failed to decode request: " <> err+        Left err@(DecodingError _) -> do+          putErrText $ toS $ displayException err           loop         Left err -> throwIO err +serverBye :: Socket.Socket -> IO ()+serverBye sock =+  Socket.LBS.sendAll sock (Aeson.encode DaemonBye) `catchAny` (\_ -> return ())+ -- | Try to read and decode a push request. readPushRequest :: Socket.Socket -> ExceptT ListenError IO (ClientMessage, Socket.Socket) readPushRequest sock = do@@ -96,8 +99,8 @@   return $ fromMaybe cacheFallback xdgRuntimeDir  -- TODO: lock the socket-openSocket :: FilePath -> IO Socket.Socket-openSocket socketFilePath = do+openSocket :: (MonadIO m) => FilePath -> m Socket.Socket+openSocket socketFilePath = liftIO $ do   deleteSocketFileIfExists socketFilePath   sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol   Socket.bind sock $ Socket.SockAddrUnix socketFilePath@@ -110,3 +113,6 @@     handleDoesNotExist e       | isDoesNotExistError e = return ()       | otherwise = throwIO e++closeSocket :: (MonadIO m) => Socket.Socket -> m ()+closeSocket = liftIO . Socket.close
+ src/Cachix/Client/Daemon/Log.hs view
@@ -0,0 +1,77 @@+module Cachix.Client.Daemon.Log+  ( new,+    withLogger,+    getKatipNamespace,+    getKatipContext,+    getKatipLogEnv,+    localLogEnv,+    localKatipContext,+    localKatipNamespace,+    toKatipLogLevel,+    Log.Logger (..),+    Log.LogLevel (..),+  )+where++import Cachix.Client.Daemon.Types.Log as Log+import qualified Control.Monad.Catch as E+import Data.Text.Lazy.Builder+import Katip (renderSeverity)+import qualified Katip+import qualified Katip.Format.Time as Katip.Format+import Katip.Scribes.Handle (brackets, colorBySeverity)+import Protolude++new :: (MonadIO m) => Katip.Namespace -> Maybe Handle -> LogLevel -> m Logger+new logLabel logHandle logLevel = do+  logKLogEnv <- liftIO $ Katip.initLogEnv logLabel ""+  let logKNamespace = mempty+  let logKContext = mempty+  return $ Logger {..}++withLogger :: (MonadIO m, E.MonadMask m) => Logger -> (Logger -> m a) -> m a+withLogger logger@(Logger {..}) f = do+  let kLogLevel = toKatipLogLevel logLevel+  let kLogHandle = fromMaybe stdout logHandle+  let registerScribe = liftIO $ do+        scribeHandle <- Katip.mkHandleScribeWithFormatter conciseBracketFormat Katip.ColorIfTerminal kLogHandle (Katip.permitItem kLogLevel) Katip.V2+        Katip.registerScribe "stdout" scribeHandle Katip.defaultScribeSettings logKLogEnv++  E.bracket registerScribe (liftIO . Katip.closeScribes) $ \logEnv ->+    f logger {logKLogEnv = logEnv}++getKatipNamespace :: Logger -> Katip.Namespace+getKatipNamespace = logKNamespace++getKatipContext :: Logger -> Katip.LogContexts+getKatipContext = logKContext++getKatipLogEnv :: Logger -> Katip.LogEnv+getKatipLogEnv = logKLogEnv++localLogEnv :: (Katip.LogEnv -> Katip.LogEnv) -> Logger -> Logger+localLogEnv f logger = logger {logKLogEnv = f (logKLogEnv logger)}++localKatipContext :: (Katip.LogContexts -> Katip.LogContexts) -> Logger -> Logger+localKatipContext f logger = logger {logKContext = f (logKContext logger)}++localKatipNamespace :: (Katip.Namespace -> Katip.Namespace) -> Logger -> Logger+localKatipNamespace f logger = logger {logKNamespace = f (logKNamespace logger)}++toKatipLogLevel :: LogLevel -> Katip.Severity+toKatipLogLevel = \case+  Debug -> Katip.DebugS+  Info -> Katip.InfoS+  Warning -> Katip.WarningS+  Error -> Katip.ErrorS++conciseBracketFormat :: (Katip.LogItem a) => Katip.ItemFormatter a+conciseBracketFormat withColor _verbosity Katip.Item {..} =+  brackets nowStr+    <> brackets (fromText (renderSeverity' _itemSeverity))+    <> fromText " "+    <> Katip.unLogStr _itemMessage+  where+    nowStr = fromText (Katip.Format.formatAsLogTime _itemTime)+    renderSeverity' severity =+      colorBySeverity withColor severity (renderSeverity severity)
+ src/Cachix/Client/Daemon/PostBuildHook.hs view
@@ -0,0 +1,120 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TypeApplications #-}++module Cachix.Client.Daemon.PostBuildHook where++import Data.Containers.ListUtils (nubOrd)+import Data.String (String)+import Data.String.Here+import Protolude+import System.Directory+  ( XdgDirectory (XdgConfig),+    XdgDirectoryList (XdgConfigDirs),+    getXdgDirectory,+    getXdgDirectoryList,+  )+import System.Environment (getExecutablePath, lookupEnv)+import System.FilePath ((</>))+import System.IO.Temp (withSystemTempDirectory)+import System.Posix.Files++type EnvVar = (String, String)++modifyEnv :: EnvVar -> [EnvVar] -> [EnvVar]+modifyEnv (envName, envValue) processEnv =+  nubOrd $ (envName, envValue) : processEnv++withSetup :: Maybe FilePath -> (FilePath -> EnvVar -> IO a) -> IO a+withSetup mdaemonSock f =+  withSystemTempDirectory "cachix-daemon" $ \tempDir -> do+    let postBuildHookScriptPath = tempDir </> "post-build-hook.sh"+        postBuildHookConfigPath = tempDir </> "nix.conf"+        daemonSock = fromMaybe (tempDir </> "daemon.sock") mdaemonSock++    cachixBin <- getExecutablePath+    writeFile postBuildHookScriptPath (postBuildHookScript cachixBin daemonSock)+    setFileMode postBuildHookScriptPath 0o755++    mnixConfEnv <- buildNixConfEnv postBuildHookScriptPath+    nixUserConfFilesEnv <- buildNixUserConfFilesEnv postBuildHookConfigPath+    nixConfEnvVar <- case mnixConfEnv of+      Just nixConfEnv -> return nixConfEnv+      Nothing -> do+        writeFile postBuildHookConfigPath (postBuildHookConfig postBuildHookScriptPath)+        return nixUserConfFilesEnv++    f daemonSock nixConfEnvVar++-- | Build the NIX_CONF environment variable.+--+-- NIX_CONF completely overrides the nix.conf.+-- This is generally undesirable because the user and system nix.confs contain important settings, like substituters.+-- Therefore, this returns Nothing if NIX_CONF is not already set to allow fallback to NIX_USER_CONF_FILES.+buildNixConfEnv :: FilePath -> IO (Maybe EnvVar)+buildNixConfEnv postBuildHookScriptPath =+  fmap appendNixConf <$> lookupEnv "NIX_CONF"+  where+    appendNixConf :: String -> EnvVar+    appendNixConf conf =+      ( "NIX_CONF",+        conf <> "\n" <> toS (postBuildHookConfig postBuildHookScriptPath)+      )++-- | Build the NIX_USER_CONF_FILES environment variable.+--+-- From man nix.conf:+--+-- If NIX_USER_CONF_FILES is set, then each path separated by : will be loaded in reverse order.+--+-- Otherwise it will look for nix/nix.conf files in XDG_CONFIG_DIRS and XDG_CONFIG_HOME. If+-- unset, XDG_CONFIG_DIRS defaults to /etc/xdg, and XDG_CONFIG_HOME defaults to $HOME/.config+-- as per XDG Base Directory Specification.+--+-- We don't need to load the system config from $NIX_CONF_DIR/nix.conf.+-- Nix loads it by default and uses it as the base config.+buildNixUserConfFilesEnv :: FilePath -> IO EnvVar+buildNixUserConfFilesEnv nixConfPath = do+  -- A user can set NIX_USER_CONF_FILES to override the default nix.conf files.+  -- In that case, we reuse it and prepend our own config file.+  mexistingEnv <- lookupEnv "NIX_USER_CONF_FILES"++  newNixUserConfFiles <- case mexistingEnv of+    Just existingEnv -> return $ nixConfPath <> ":" <> existingEnv+    Nothing -> do+      userConfigFiles <- getUserConfigFiles++      -- Combine all the nix.conf paths into one string, separated by colons.+      -- Filter out empty paths.+      return $ intercalate ":" $ filter (not . null) $ nixConfPath : userConfigFiles++  return ("NIX_USER_CONF_FILES", newNixUserConfFiles)++getUserConfigFiles :: IO [FilePath]+getUserConfigFiles =+  fmap (</> "nix/nix.conf") <$> getUserConfigDirs++getUserConfigDirs :: IO [FilePath]+getUserConfigDirs = do+  configHome <- getXdgDirectory XdgConfig empty+  configDirs <- getXdgDirectoryList XdgConfigDirs++  return $ configHome : configDirs++postBuildHookConfig :: FilePath -> Text+postBuildHookConfig scriptPath =+  [iTrim|+post-build-hook = ${toS scriptPath :: Text}+  |]++postBuildHookScript :: FilePath -> FilePath -> Text+postBuildHookScript cachixBin socketPath =+  [iTrim|+\#!/bin/sh++\# set -eu+set -f # disable globbing++exec ${toS cachixBin :: Text} daemon push \\+  --socket ${toS socketPath :: Text} \\+  $OUT_PATHS+  |]
+ src/Cachix/Client/Daemon/Progress.hs view
@@ -0,0 +1,166 @@+module Cachix.Client.Daemon.Progress+  ( UploadProgress,+    new,+    complete,+    fail,+    tick,+    update,+  )+where++import Cachix.Client.Daemon.Types (PushRetryStatus (..))+import Cachix.Client.HumanSize (humanSize)+import qualified Control.Concurrent.Async as Async+import Control.Concurrent.MVar+import Data.String (String)+import qualified Data.Text as T+import Protolude+import qualified System.Console.AsciiProgress as Ascii+import qualified System.Console.AsciiProgress.Internal as Ascii.Internal+import System.Console.Pretty+import System.IO (hIsTerminalDevice)++data UploadProgress+  = ProgressBar+      { path :: String,+        size :: Int64,+        progressBar :: Ascii.ProgressBar+      }+  | FallbackText+      { path :: String,+        size :: Int64+      }++new :: Handle -> String -> Int64 -> PushRetryStatus -> IO UploadProgress+new hdl path size retryStatus = do+  isTerminal <- liftIO $ hIsTerminalDevice hdl+  if isTerminal+    then do+      progressBar <- newProgressBar path size retryStatus+      return $ ProgressBar {..}+    else do+      hPutStr stderr $ uploadStartFallback path size retryStatus+      return $ FallbackText {..}++complete :: UploadProgress -> IO ()+complete ProgressBar {..} = Ascii.complete progressBar+complete _ = pure ()++tick :: UploadProgress -> Int64 -> IO ()+tick ProgressBar {..} deltaBytes = Ascii.tickN progressBar (fromIntegral deltaBytes)+tick _ _ = pure ()++fail :: UploadProgress -> IO ()+fail ProgressBar {..} =+  clearAsciiWith progressBar (uploadFailed path)+fail FallbackText {path} =+  hPutStr stderr $ uploadFailed path++update :: UploadProgress -> PushRetryStatus -> IO UploadProgress+update pg@ProgressBar {..} retryStatus = do+  let opts = newProgressBarOptions path size retryStatus+  pg' <- newAsciiProgressBarInPlace progressBar opts++  return $ pg {progressBar = pg'}+update fbt _ = return fbt++newProgressBar :: String -> Int64 -> PushRetryStatus -> IO Ascii.ProgressBar+newProgressBar path size retryStatus =+  newAsciiProgressBar $ newProgressBarOptions path size retryStatus++newProgressBarOptions :: String -> Int64 -> PushRetryStatus -> Ascii.Options+newProgressBarOptions path size retryStatus = do+  let hSize = toS $ humanSize $ fromIntegral size+  let barLength =+        uploadTickBar path hSize retryStatus+          & toS+          & T.replace "[:bar]" ""+          & T.replace ":percent" "  0%"+          & T.length+  Ascii.def+    { Ascii.pgTotal = fromIntegral size,+      -- https://github.com/yamadapc/haskell-ascii-progress/issues/24+      Ascii.pgWidth = 20 + barLength,+      Ascii.pgOnCompletion = Just $ uploadComplete path size,+      Ascii.pgFormat = uploadTickBar path hSize retryStatus+    }++retryText :: PushRetryStatus -> String+retryText PushRetryStatus {retryCount} =+  if retryCount == 0+    then ""+    else color Yellow $ "retry #" <> show retryCount <> " "++uploadComplete :: String -> Int64 -> String+uploadComplete path size =+  let hSize = toS $ humanSize $ fromIntegral size+   in color Green "✓ " <> path <> " (" <> hSize <> ")"++uploadFailed :: String -> String+uploadFailed path =+  color Red "✗ " <> path++uploadTickBar :: String -> String -> PushRetryStatus -> String+uploadTickBar path hSize retryStatus =+  color Blue "[:bar] " <> retryText retryStatus <> toS path <> " (:percent of " <> hSize <> ")"++uploadStartFallback :: String -> Int64 -> PushRetryStatus -> String+uploadStartFallback path size retryStatus =+  let hSize = toS $ humanSize $ fromIntegral size+   in retryText retryStatus <> "Pushing " <> path <> " (" <> hSize <> ")\n"++-- Internal++newAsciiProgressBar :: Ascii.Options -> IO Ascii.ProgressBar+newAsciiProgressBar opts = do+  region <- Ascii.openConsoleRegion Ascii.Linear+  newAsciiProgressBarWithRegion opts region++-- | Create a new progress bar in place of an existing one, reusing the console region.+newAsciiProgressBarInPlace :: Ascii.ProgressBar -> Ascii.Options -> IO Ascii.ProgressBar+newAsciiProgressBarInPlace pg opts = do+  cancelUpdates pg+  newAsciiProgressBarWithRegion opts (Ascii.pgRegion pg)++-- | Create a new progress bar using the provided console region.+newAsciiProgressBarWithRegion :: Ascii.Options -> Ascii.ConsoleRegion -> IO Ascii.ProgressBar+newAsciiProgressBarWithRegion opts region = do+  info <- Ascii.Internal.newProgressBarInfo opts++  -- Display initial progress-bar+  pgStr <- Ascii.pgGetProgressStr opts opts <$> Ascii.Internal.getInfoStats info+  Ascii.setConsoleRegion region pgStr++  future <- Async.async $ start info+  return $ Ascii.ProgressBar info future region+  where+    start info@Ascii.Internal.ProgressBarInfo {..} = do+      c <- readMVar pgCompleted+      unlessDone c $ do+        n <- readChan pgChannel+        _ <- handleMessage info n+        unlessDone (c + n) $ start info+      where+        unlessDone c action | c < Ascii.pgTotal opts = action+        unlessDone _ _ = do+          let fmt = fromMaybe (Ascii.pgFormat opts) (Ascii.pgOnCompletion opts)+          onCompletion <- Ascii.pgGetProgressStr opts opts {Ascii.pgFormat = fmt} <$> Ascii.Internal.getInfoStats info+          Ascii.setConsoleRegion region onCompletion++    handleMessage info n = do+      -- Update the completed tick count+      modifyMVar_ (Ascii.Internal.pgCompleted info) (\c -> return (c + n))+      -- Find and update the current and first tick times:+      stats <- Ascii.Internal.getInfoStats info+      let progressStr = Ascii.Internal.pgGetProgressStr opts opts stats+      Ascii.setConsoleRegion region progressStr++-- | Cancel async updates.+cancelUpdates :: Ascii.ProgressBar -> IO ()+cancelUpdates (Ascii.ProgressBar _ future _) = Async.cancel future++-- | Cancel async updates and clear the console region with the given string.+clearAsciiWith :: Ascii.ProgressBar -> String -> IO ()+clearAsciiWith pg@(Ascii.ProgressBar _ _ region) str = do+  cancelUpdates pg+  Ascii.setConsoleRegion region str
+ src/Cachix/Client/Daemon/Protocol.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Cachix.Client.Daemon.Protocol+  ( ClientMessage (..),+    DaemonMessage (..),+    PushRequestId,+    newPushRequestId,+    PushRequest (..),+  )+where++import qualified Data.Aeson as Aeson+import Data.UUID (UUID)+import qualified Data.UUID.V4 as UUID+import Protolude++-- | JSON messages that the client can send to the daemon+data ClientMessage+  = ClientPushRequest PushRequest+  | ClientStop+  deriving stock (Generic)+  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)++-- | JSON messages that the daemon can send to the client+data DaemonMessage+  = DaemonBye+  deriving stock (Generic)+  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)++newtype PushRequestId = PushRequestId UUID+  deriving stock (Generic)+  deriving newtype (Eq, Ord, Show, Aeson.FromJSON, Aeson.ToJSON, Hashable)++newPushRequestId :: (MonadIO m) => m PushRequestId+newPushRequestId = liftIO $ PushRequestId <$> UUID.nextRandom++-- | A request for the daemon to push store paths to a binary cache+data PushRequest = PushRequest+  { storePaths :: [FilePath]+  }+  deriving stock (Eq, Generic, Show)+  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)
+ src/Cachix/Client/Daemon/Push.hs view
@@ -0,0 +1,61 @@+module Cachix.Client.Daemon.Push where++import qualified Cachix.API as API+import Cachix.Client.Commands.Push hiding (pushStrategy)+import qualified Cachix.Client.Daemon.PushManager as PushManager+import Cachix.Client.Daemon.Types (Daemon, DaemonEnv (..), PushManager)+import Cachix.Client.Env (Env (..))+import Cachix.Client.OptionsParser as Client.OptionsParser+  ( PushOptions (..),+  )+import Cachix.Client.Push as Client.Push+import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Servant+import Cachix.Types.BinaryCache (BinaryCacheName)+import qualified Cachix.Types.BinaryCache as BinaryCache+import qualified Data.Set as Set+import Hercules.CNix.Store (withStore)+import Protolude hiding (toS)+import Servant.Auth ()+import Servant.Auth.Client+import Servant.Client.Streaming+import Servant.Conduit ()++withPushParams :: (PushParams PushManager () -> Daemon b) -> Daemon b+withPushParams m = do+  DaemonEnv {..} <- ask+  let authToken = getAuthTokenFromPushSecret daemonPushSecret+      cacheName = BinaryCache.name daemonBinaryCache+      compressionMethod = getCompressionMethod daemonPushOptions daemonBinaryCache++  withStore $ \store ->+    m $ do+      let pushStrategy = PushManager.newPushStrategy store authToken daemonPushOptions cacheName compressionMethod++      PushParams+        { pushParamsName = cacheName,+          pushParamsSecret = daemonPushSecret,+          pushParamsClientEnv = clientenv daemonEnv,+          pushOnClosureAttempt = \full missing -> do+            let already = Set.toList $ Set.difference (Set.fromList full) (Set.fromList missing)+            mapM_ (onAlreadyPresent . pushStrategy) already+            return missing,+          pushParamsStrategy = pushStrategy,+          pushParamsStore = store+        }++getBinaryCache :: Env -> Maybe Token -> BinaryCacheName -> IO BinaryCache.BinaryCache+getBinaryCache env authToken name = do+  -- Self-signed caches might not have a token, which is why this code is so weird.+  -- In practice, public self-signed caches don't need one and private ones always need a token.+  let token = fromMaybe (Token "") authToken+  res <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token name+  case res of+    Left err -> handleCacheResponse name authToken err+    Right binaryCache -> pure binaryCache++getCompressionMethod :: PushOptions -> BinaryCache.BinaryCache -> BinaryCache.CompressionMethod+getCompressionMethod opts binaryCache =+  fromMaybe BinaryCache.ZSTD $+    Client.OptionsParser.compressionMethod opts+      <|> Just (BinaryCache.preferredCompressionMethod binaryCache)
+ src/Cachix/Client/Daemon/PushManager.hs view
@@ -0,0 +1,386 @@+module Cachix.Client.Daemon.PushManager+  ( newPushManagerEnv,+    runPushManager,+    stopPushManager,++    -- * Push strategy+    newPushStrategy,++    -- * Push job+    PushJob (..),+    addPushJob,+    lookupPushJob,+    withPushJob,+    resolvePushJob,++    -- * Store paths+    queueStorePaths,+    removeStorePath,+    queuedStorePathCount,++    -- * Tasks+    handleTask,+    -- Push events+    pushStarted,+    pushFinished,+    pushStorePathAttempt,+    pushStorePathProgress,+    pushStorePathDone,+    pushStorePathFailed,+  )+where++import Cachix.Client.CNix (filterInvalidStorePath, followLinksToStorePath)+import Cachix.Client.Commands.Push hiding (pushStrategy)+import qualified Cachix.Client.Daemon.Protocol as Protocol+import qualified Cachix.Client.Daemon.PushManager.PushJob as PushJob+import Cachix.Client.Daemon.Types.Log (Logger)+import Cachix.Client.Daemon.Types.PushEvent+import Cachix.Client.Daemon.Types.PushManager+import Cachix.Client.OptionsParser as Client.OptionsParser+  ( PushOptions (..),+  )+import Cachix.Client.Push as Client.Push+import Cachix.Client.Retry (retryAll)+import qualified Cachix.Types.BinaryCache as BinaryCache+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)+import qualified Data.ByteString as BS+import qualified Data.HashMap.Strict as HashMap+import Data.IORef+import qualified Data.Set as Set+import qualified Data.Text as T+import Data.Time (getCurrentTime)+import Hercules.CNix (StorePath)+import Hercules.CNix.Store (Store, parseStorePath, storePathToPath)+import qualified Katip+import Protolude hiding (toS)+import Protolude.Conv (toS)+import Servant.Auth ()+import Servant.Auth.Client+import Servant.Conduit ()+import qualified UnliftIO.QSem as QSem++newPushManagerEnv :: (MonadIO m) => PushOptions -> Logger -> OnPushEvent -> m PushManagerEnv+newPushManagerEnv pushOptions pmLogger pmOnPushEvent = liftIO $ do+  pmPushJobs <- newTVarIO mempty+  pmStorePathReferences <- newTVarIO mempty+  pmTaskQueue <- newTBMQueueIO 1000+  pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)+  return $ PushManagerEnv {..}++runPushManager :: (MonadIO m) => PushManagerEnv -> PushManager a -> m a+runPushManager env f = liftIO $ unPushManager f `runReaderT` env++stopPushManager :: PushManagerEnv -> IO ()+stopPushManager PushManagerEnv {pmTaskQueue, pmPushJobs} =+  atomically $ do+    pushJobs <- readTVar pmPushJobs+    if all PushJob.isCompleted (HashMap.elems pushJobs)+      then closeTBMQueue pmTaskQueue+      else retry++-- Manage push jobs++addPushJob :: Protocol.PushRequest -> PushManager Protocol.PushRequestId+addPushJob pushRequest = do+  PushManagerEnv {..} <- ask+  pushJob <- PushJob.new pushRequest++  Katip.logLocM Katip.DebugS $ Katip.ls $ "Queued push job " <> (show (pushId pushJob) :: Text)++  liftIO $+    atomically $ do+      modifyTVar' pmPushJobs $ HashMap.insert (pushId pushJob) pushJob+      writeTBMQueue pmTaskQueue $ ResolveClosure (pushId pushJob)++  return (pushId pushJob)++removePushJob :: Protocol.PushRequestId -> PushManager ()+removePushJob pushId = do+  PushManagerEnv {..} <- ask+  liftIO $ atomically $ modifyTVar' pmPushJobs $ HashMap.delete pushId++lookupPushJob :: Protocol.PushRequestId -> PushManager (Maybe PushJob)+lookupPushJob pushId = do+  pushJobs <- asks pmPushJobs+  liftIO $ HashMap.lookup pushId <$> readTVarIO pushJobs++withPushJob :: Protocol.PushRequestId -> (PushJob -> PushManager ()) -> PushManager ()+withPushJob pushId f =+  maybe handleMissingPushJob f =<< lookupPushJob pushId+  where+    handleMissingPushJob =+      Katip.logLocM Katip.ErrorS $ Katip.ls $ "Push job " <> (show pushId :: Text) <> " not found"++modifyPushJob :: Protocol.PushRequestId -> (PushJob -> PushJob) -> PushManager (Maybe PushJob)+modifyPushJob pushId f = do+  pushJobs <- asks pmPushJobs+  liftIO $ atomically $ stateTVar pushJobs $ \jobs -> do+    let pj = HashMap.adjust f pushId jobs+    (HashMap.lookup pushId pj, pj)++modifyPushJobs :: [Protocol.PushRequestId] -> (PushJob -> PushJob) -> PushManager ()+modifyPushJobs pushIds f = do+  pushJobs <- asks pmPushJobs+  liftIO $ atomically $ modifyTVar' pushJobs $ \pushJobs' ->+    foldl' (flip (HashMap.adjust f)) pushJobs' pushIds++-- Manage store paths++queueStorePaths :: Protocol.PushRequestId -> [FilePath] -> PushManager ()+queueStorePaths pushId storePaths = do+  PushManagerEnv {..} <- ask++  let addToQueue storePath = do+        isDuplicate <- HashMap.member storePath <$> readTVar pmStorePathReferences+        unless isDuplicate $+          writeTBMQueue pmTaskQueue (PushStorePath storePath)++        modifyTVar' pmStorePathReferences $ HashMap.insertWith (<>) storePath [pushId]++  transactionally $ map addToQueue storePaths++removeStorePath :: FilePath -> PushManager ()+removeStorePath storePath = do+  pmStorePathReferences <- asks pmStorePathReferences+  liftIO $ atomically $ do+    modifyTVar' pmStorePathReferences $ HashMap.delete storePath++lookupStorePathReferences :: FilePath -> PushManager [Protocol.PushRequestId]+lookupStorePathReferences storePath = do+  pmStorePathReferences <- asks pmStorePathReferences+  references <- liftIO $ readTVarIO pmStorePathReferences+  return $ fromMaybe [] (HashMap.lookup storePath references)++checkPushJobCompleted :: Protocol.PushRequestId -> PushManager ()+checkPushJobCompleted pushId = do+  mpushJob <- lookupPushJob pushId+  for_ mpushJob $ \pushJob ->+    when (Set.null $ pushQueue pushJob) $ do+      timestamp <- liftIO getCurrentTime+      _ <- modifyPushJob pushId $ PushJob.complete timestamp+      pushFinished pushJob++queuedStorePathCount :: PushManager Integer+queuedStorePathCount = do+  pmPushJobs <- asks pmPushJobs+  jobs <- liftIO $ readTVarIO pmPushJobs+  pure $ foldl' countQueuedPaths 0 (HashMap.elems jobs)+  where+    countQueuedPaths acc job = acc + fromIntegral (Set.size $ pushQueue job)++resolvePushJob :: Protocol.PushRequestId -> PushJob.ResolvedClosure FilePath -> PushManager ()+resolvePushJob pushId closure = do+  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)+    -- Check if the job is already completed, i.e. all paths have been skipped.+    checkPushJobCompleted pushId+  where+    showClosureStats :: PushJob.ResolvedClosure FilePath -> Text+    showClosureStats PushJob.ResolvedClosure {..} =+      let skippedPaths = Set.difference rcAllPaths rcMissingPaths+          queuedCount = length rcMissingPaths+          skippedCount = length skippedPaths+          totalCount = queuedCount + skippedCount+       in T.intercalate+            "\n"+            [ "Resolved push job " <> show pushId,+              "Total paths: " <> show totalCount,+              "Queued paths: " <> show queuedCount,+              "Skipped paths: " <> show skippedCount+            ]++handleTask :: PushParams PushManager () -> Task -> PushManager ()+handleTask pushParams task = do+  case task of+    ResolveClosure pushId -> do+      Katip.logLocM Katip.DebugS $ Katip.ls $ "Resolving closure for push job " <> (show pushId :: Text)++      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+              }++        resolvePushJob pushId resolvedClosure+    PushStorePath filePath -> do+      qs <- asks pmTaskSemaphore+      E.bracket_ (QSem.waitQSem qs) (QSem.signalQSem qs) $ do+        Katip.logLocM Katip.DebugS $ Katip.ls $ "Pushing store path " <> filePath++        let store = pushParamsStore pushParams+        storePath <- liftIO $ parseStorePath store (toS filePath)++        retryAll (uploadStorePath pushParams storePath)+          `Safe.catchAny` (pushStorePathFailed filePath . toS . displayException)++newPushStrategy ::+  Store ->+  Maybe Token ->+  PushOptions ->+  Text ->+  BinaryCache.CompressionMethod ->+  (StorePath -> PushStrategy PushManager ())+newPushStrategy store authToken opts cacheName compressionMethod storePath =+  let onAlreadyPresent = do+        sp <- liftIO $ storePathToPath store storePath+        Katip.logFM Katip.InfoS $ Katip.ls $ "Skipping " <> (toS sp :: Text)+        -- TODO: needs another event type here+        pushStorePathDone (toS sp)++      onError err = do+        let errText = toS (displayException err)+        sp <- liftIO $ storePathToPath store storePath+        Katip.katipAddContext (Katip.sl "error" errText) $+          Katip.logFM Katip.InfoS (Katip.ls $ "Failed " <> (toS sp :: Text))+        pushStorePathFailed (toS sp) errText++      onAttempt retryStatus size = do+        sp <- liftIO $ storePathToPath store storePath+        Katip.logFM Katip.InfoS $ Katip.ls $ "Pushing " <> (toS sp :: Text)+        pushStorePathAttempt (toS sp) size retryStatus++      onUncompressedNARStream _ size = do+        sp <- liftIO $ storePathToPath store storePath+        lastEmitRef <- liftIO $ newIORef (0 :: Int64)+        currentBytesRef <- liftIO $ newIORef (0 :: Int64)+        C.awaitForever $ \chunk -> do+          let newBytes = fromIntegral (BS.length chunk)+          currentBytes <- liftIO $ atomicModifyIORef' currentBytesRef (\b -> (b + newBytes, b + newBytes))+          lastEmit <- liftIO $ readIORef lastEmitRef++          when (currentBytes - lastEmit >= 1024 || currentBytes == size) $ do+            liftIO $ writeIORef lastEmitRef currentBytes+            lift $ lift $ pushStorePathProgress (toS sp) currentBytes newBytes++          C.yield chunk++      onDone = do+        sp <- liftIO $ storePathToPath store storePath+        Katip.logFM Katip.InfoS $ Katip.ls $ "Pushed " <> (toS sp :: Text)+        pushStorePathDone (toS sp)+   in PushStrategy+        { onAlreadyPresent = onAlreadyPresent,+          on401 = liftIO . handleCacheResponse cacheName authToken,+          onError = onError,+          onAttempt = onAttempt,+          onUncompressedNARStream = onUncompressedNARStream,+          onDone = onDone,+          Client.Push.compressionMethod = compressionMethod,+          Client.Push.compressionLevel = Client.OptionsParser.compressionLevel opts,+          Client.Push.omitDeriver = Client.OptionsParser.omitDeriver opts+        }++-- Push events++pushStarted :: PushJob -> PushManager ()+pushStarted pushJob@PushJob {pushId} = do+  case PushJob.startedAt pushJob of+    Nothing -> return ()+    Just timestamp -> do+      sendPushEvent <- asks pmOnPushEvent+      liftIO $ do+        sendPushEvent pushId $+          PushEvent timestamp pushId PushStarted++pushFinished :: PushJob -> PushManager ()+pushFinished pushJob@PushJob {pushId} = void $ runMaybeT $ do+  pushDuration <- MaybeT $ pure $ PushJob.duration pushJob+  completedAt <- MaybeT $ pure $ PushJob.completedAt pushJob++  Katip.logLocM Katip.InfoS $+    Katip.ls $+      T.intercalate+        " "+        [ "Push job",+          show pushId :: Text,+          "finished in",+          show pushDuration+        ]++  sendPushEvent <- asks pmOnPushEvent+  liftIO $ do+    sendPushEvent pushId $+      PushEvent completedAt pushId PushFinished++  lift $ removePushJob pushId++sendStorePathEvent :: [Protocol.PushRequestId] -> PushEventMessage -> PushManager ()+sendStorePathEvent pushIds msg = do+  timestamp <- liftIO getCurrentTime+  sendPushEvent <- asks pmOnPushEvent+  liftIO $ forM_ pushIds $ \pushId ->+    sendPushEvent pushId (PushEvent timestamp pushId msg)++pushStorePathAttempt :: FilePath -> Int64 -> RetryStatus -> PushManager ()+pushStorePathAttempt storePath size retryStatus = do+  let pushRetryStatus = newPushRetryStatus retryStatus+  pushIds <- lookupStorePathReferences storePath+  sendStorePathEvent pushIds (PushStorePathAttempt storePath size pushRetryStatus)++pushStorePathProgress :: FilePath -> Int64 -> Int64 -> PushManager ()+pushStorePathProgress storePath currentBytes newBytes = do+  pushIds <- lookupStorePathReferences storePath+  sendStorePathEvent pushIds (PushStorePathProgress storePath currentBytes newBytes)++pushStorePathDone :: FilePath -> PushManager ()+pushStorePathDone storePath = do+  pushIds <- lookupStorePathReferences storePath+  modifyPushJobs pushIds (PushJob.markStorePathPushed storePath)++  sendStorePathEvent pushIds (PushStorePathDone storePath)++  forM_ pushIds checkPushJobCompleted++  removeStorePath storePath++pushStorePathFailed :: FilePath -> Text -> PushManager ()+pushStorePathFailed storePath errMsg = do+  pushIds <- lookupStorePathReferences storePath+  modifyPushJobs pushIds (PushJob.markStorePathFailed storePath)++  sendStorePathEvent pushIds (PushStorePathFailed storePath errMsg)++  forM_ pushIds checkPushJobCompleted++  removeStorePath storePath++-- Helpers++transactionally :: (Foldable t, MonadIO m) => t (STM ()) -> m ()+transactionally = liftIO . atomically . sequence_++storeToFilePath :: (MonadIO m) => Store -> StorePath -> m FilePath+storeToFilePath store storePath = do+  fp <- liftIO $ storePathToPath store storePath+  pure $ toS fp++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
+ src/Cachix/Client/Daemon/PushManager/PushJob.hs view
@@ -0,0 +1,106 @@+module Cachix.Client.Daemon.PushManager.PushJob+  ( module Cachix.Client.Daemon.PushManager.PushJob,+    module Types,+  )+where++import qualified Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.Types.PushManager as Types+import qualified Data.Set as Set+import Data.Time (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime)+import Protolude++new :: (MonadIO m) => Protocol.PushRequest -> m PushJob+new pushRequest = do+  pushId <- Protocol.newPushRequestId+  timestamp <- liftIO getCurrentTime+  return $+    PushJob+      { pushId,+        pushRequest,+        pushStatus = Queued,+        pushQueue = mempty,+        pushResult = mempty,+        pushStats = newStats timestamp+      }++newStats :: UTCTime -> JobStats+newStats createdAt =+  JobStats+    { jsCreatedAt = createdAt,+      jsStartedAt = Nothing,+      jsCompletedAt = Nothing+    }++data ResolvedClosure p = ResolvedClosure+  { rcAllPaths :: Set p,+    rcMissingPaths :: Set p+  }++populateQueue :: ResolvedClosure FilePath -> UTCTime -> PushJob -> PushJob+populateQueue ResolvedClosure {..} timestamp pushJob@PushJob {..} = do+  let skippedPaths = Set.difference rcAllPaths rcMissingPaths+  pushJob+    { pushStatus = Running,+      pushStats = pushStats {jsStartedAt = Just timestamp},+      pushQueue = rcMissingPaths,+      pushResult = pushResult {prSkippedPaths = skippedPaths}+    }++addPushedPath :: FilePath -> PushResult -> PushResult+addPushedPath storePath pushResult =+  pushResult {prPushedPaths = Set.insert storePath (prPushedPaths pushResult)}++addFailedPath :: FilePath -> PushResult -> PushResult+addFailedPath storePath pushResult =+  pushResult {prFailedPaths = Set.insert storePath (prFailedPaths pushResult)}++markStorePathPushed :: FilePath -> PushJob -> PushJob+markStorePathPushed storePath pushJob@(PushJob {pushQueue, pushResult}) =+  pushJob+    { pushStatus = Running,+      pushQueue = Set.delete storePath pushQueue,+      pushResult = addPushedPath storePath pushResult+    }++markStorePathFailed :: FilePath -> PushJob -> PushJob+markStorePathFailed storePath pushJob@(PushJob {pushQueue, pushResult}) =+  pushJob+    { pushStatus = Running,+      pushQueue = Set.delete storePath pushQueue,+      pushResult = addFailedPath storePath pushResult+    }++status :: PushJob -> JobStatus+status PushJob {pushStatus} = pushStatus++queue :: PushJob -> Set FilePath+queue PushJob {pushQueue} = pushQueue++result :: PushJob -> PushResult+result PushJob {pushResult} = pushResult++hasQueuedPaths :: PushJob -> Bool+hasQueuedPaths = not . Set.null . queue++complete :: UTCTime -> PushJob -> PushJob+complete timestamp pushJob@PushJob {..} = do+  pushJob+    { pushStatus = Completed,+      pushStats = pushStats {jsCompletedAt = Just timestamp}+    }++isCompleted :: PushJob -> Bool+isCompleted PushJob {pushStatus} = pushStatus == Completed++startedAt :: PushJob -> Maybe UTCTime+startedAt PushJob {pushStats = JobStats {jsStartedAt}} = jsStartedAt++completedAt :: PushJob -> Maybe UTCTime+completedAt PushJob {pushStats = JobStats {jsCompletedAt}} = jsCompletedAt++duration :: PushJob -> Maybe NominalDiffTime+duration PushJob {pushStats} = do+  t1 <- jsStartedAt pushStats+  t2 <- jsCompletedAt pushStats+  pure $ diffUTCTime t2 t1
+ src/Cachix/Client/Daemon/ShutdownLatch.hs view
@@ -0,0 +1,26 @@+module Cachix.Client.Daemon.ShutdownLatch+  ( ShutdownLatch,+    newShutdownLatch,+    waitForShutdown,+    initiateShutdown,+    isShuttingDown,+  )+where++import Control.Concurrent.MVar+import Protolude++-- | A latch to keep track of the shutdown process.+newtype ShutdownLatch = ShutdownLatch {unShutdownLatch :: MVar ()}++newShutdownLatch :: (MonadIO m) => m ShutdownLatch+newShutdownLatch = ShutdownLatch <$> liftIO newEmptyMVar++waitForShutdown :: (MonadIO m) => ShutdownLatch -> m ()+waitForShutdown = liftIO . readMVar . unShutdownLatch++initiateShutdown :: (MonadIO m) => ShutdownLatch -> m ()+initiateShutdown = void . liftIO . flip tryPutMVar () . unShutdownLatch++isShuttingDown :: (MonadIO m) => ShutdownLatch -> m Bool+isShuttingDown = liftIO . fmap not . isEmptyMVar . unShutdownLatch
+ src/Cachix/Client/Daemon/Subscription.hs view
@@ -0,0 +1,102 @@+module Cachix.Client.Daemon.Subscription where++import Control.Concurrent.STM.TBMQueue+import Control.Concurrent.STM.TMChan+import Control.Concurrent.STM.TVar+import Data.Aeson as Aeson (ToJSON)+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HashMap+import qualified Network.Socket as Socket+import Protolude++data SubscriptionManager k v = SubscriptionManager+  { managerSubscriptions :: TVar (HashMap k [Subscription v]),+    managerGlobalSubscriptions :: TVar [Subscription v],+    managerEvents :: TBMQueue (k, v)+  }++data Subscription v+  = -- | A subscriber that listens on a socket.+    SubSocket (TBMQueue v) Socket.Socket+  | -- | A subscriber that listens on a channel.+    SubChannel (TMChan v)++newSubscriptionManager :: IO (SubscriptionManager k v)+newSubscriptionManager = do+  subscriptions <- newTVarIO HashMap.empty+  globalSubscriptions <- newTVarIO []+  events <- newTBMQueueIO 10000+  pure $ SubscriptionManager subscriptions globalSubscriptions events++-- Subscriptions++subscribeTo :: (Hashable k, MonadIO m) => SubscriptionManager k v -> k -> Subscription v -> m ()+subscribeTo manager key subscription =+  liftIO $ atomically $ subscribeToSTM manager key subscription++subscribeToAll :: (MonadIO m) => SubscriptionManager k v -> Subscription v -> m ()+subscribeToAll manager subscription =+  liftIO $ atomically $ subscribeToAllSTM manager subscription++getSubscriptionsFor :: (Hashable k, MonadIO m) => SubscriptionManager k v -> k -> m [Subscription v]+getSubscriptionsFor manager key =+  liftIO $ atomically $ getSubscriptionsForSTM manager key++subscribeToSTM :: (Hashable k) => SubscriptionManager k v -> k -> Subscription v -> STM ()+subscribeToSTM manager key subscription = do+  subscriptions <- readTVar $ managerSubscriptions manager+  let subscriptions' = HashMap.insertWith (<>) key [subscription] subscriptions+  writeTVar (managerSubscriptions manager) subscriptions'++subscribeToAllSTM :: SubscriptionManager k v -> Subscription v -> STM ()+subscribeToAllSTM manager subscription = do+  subscriptions <- readTVar $ managerGlobalSubscriptions manager+  let subscriptions' = subscription : subscriptions+  writeTVar (managerGlobalSubscriptions manager) subscriptions'++getSubscriptionsForSTM :: (Hashable k) => SubscriptionManager k v -> k -> STM [Subscription v]+getSubscriptionsForSTM manager key = do+  subscriptions <- readTVar $ managerSubscriptions manager+  pure $ HashMap.lookupDefault [] key subscriptions++-- Events++pushEvent :: (MonadIO m) => SubscriptionManager k v -> k -> v -> m ()+pushEvent manager key event =+  liftIO $ atomically $ pushEventSTM manager key event++pushEventSTM :: SubscriptionManager k v -> k -> v -> STM ()+pushEventSTM manager key event =+  writeTBMQueue (managerEvents manager) (key, event)++sendEventToSub :: Subscription v -> v -> STM ()+-- TODO: implement socket subscriptions.+sendEventToSub (SubSocket _queue _) _ = pure () -- writeTBMQueue queue+sendEventToSub (SubChannel chan) event = writeTMChan chan event++runSubscriptionManager :: (Show k, Show v, Hashable k, ToJSON v, MonadIO m) => SubscriptionManager k v -> m ()+runSubscriptionManager manager = do+  isDone <- liftIO $ atomically $ do+    mevent <- readTBMQueue (managerEvents manager)+    case mevent of+      Nothing -> return True+      Just (key, event) -> do+        subscriptions <- getSubscriptionsForSTM manager key+        globalSubscriptions <- readTVar $ managerGlobalSubscriptions manager+        mapM_ (`sendEventToSub` event) (subscriptions <> globalSubscriptions)+        return False++  unless isDone $+    runSubscriptionManager manager++stopSubscriptionManager :: SubscriptionManager k v -> IO ()+stopSubscriptionManager manager = do+  liftIO $ atomically $ closeTBMQueue (managerEvents manager)+  globalSubscriptions <- liftIO $ readTVarIO $ managerGlobalSubscriptions manager+  subscriptions <- liftIO $ readTVarIO $ managerSubscriptions manager++  forM_ (concat subscriptions <> globalSubscriptions) $ \case+    SubSocket queue sock -> do+      atomically $ closeTBMQueue queue+      Socket.close sock+    SubChannel channel -> atomically $ closeTMChan channel
src/Cachix/Client/Daemon/Types.hs view
@@ -1,32 +1,26 @@-module Cachix.Client.Daemon.Types where+module Cachix.Client.Daemon.Types+  ( -- * Daemon+    DaemonEnv (..),+    Daemon,+    runDaemon, -import Cachix.Client.Config.Orphans ()-import Cachix.Client.URI-import qualified Data.Aeson as Aeson-import qualified Network.Socket as Socket-import Protolude-import Servant.Auth.Client (Token)+    -- * Log+    LogLevel (..), --- | JSON messages that the client can send to the daemon-data ClientMessage-  = ClientPushRequest PushRequest-  | ClientStop-  deriving stock (Generic)-  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)+    -- * Push+    PushManager.PushManagerEnv (..),+    PushManager.PushManager,+    PushManager.PushJob (..),+    Task (..), --- | A request for the daemon to push store paths to a binary cache-data PushRequest = PushRequest-  { authToken :: Token,-    binaryCacheName :: Text,-    host :: URI, -- TODO: host is not currently supported-    storePaths :: [FilePath]-  }-  deriving stock (Generic)-  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)+    -- * Push Event+    PushEvent.PushEvent (..),+    PushEvent.PushEventMessage (..),+    PushEvent.PushRetryStatus (..),+  )+where -data QueuedPushRequest = QueuedPushRequest-  { -- | The original push request-    pushRequest :: PushRequest,-    -- | An open socket to the client that sent the push request.-    clientConnection :: Socket.Socket-  }+import Cachix.Client.Daemon.Types.Daemon+import Cachix.Client.Daemon.Types.Log+import Cachix.Client.Daemon.Types.PushEvent as PushEvent+import Cachix.Client.Daemon.Types.PushManager as PushManager
+ src/Cachix/Client/Daemon/Types/Daemon.hs view
@@ -0,0 +1,85 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Cachix.Client.Daemon.Types.Daemon+  ( -- * Daemon+    DaemonEnv (..),+    Daemon,+    runDaemon,+  )+where++import Cachix.Client.Config.Orphans ()+import qualified Cachix.Client.Daemon.Log as Log+import qualified Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.ShutdownLatch (ShutdownLatch)+import Cachix.Client.Daemon.Subscription (SubscriptionManager)+import Cachix.Client.Daemon.Types.Log (Logger)+import Cachix.Client.Daemon.Types.PushEvent (PushEvent)+import Cachix.Client.Daemon.Types.PushManager (PushManagerEnv (..))+import Cachix.Client.Env as Env+import Cachix.Client.OptionsParser (PushOptions)+import Cachix.Client.Push+import Cachix.Types.BinaryCache (BinaryCache, BinaryCacheName)+import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow)+import Control.Monad.IO.Unlift (MonadUnliftIO)+import qualified Katip+import Protolude hiding (bracketOnError)+import System.Posix.Types (ProcessID)++data DaemonEnv = DaemonEnv+  { -- | Cachix client env+    daemonEnv :: Env,+    -- | Push options, like compression settings and number of jobs+    daemonPushOptions :: PushOptions,+    -- | Path to the socket that the daemon listens on+    daemonSocketPath :: FilePath,+    -- | The push secret for the binary cache+    daemonPushSecret :: PushSecret,+    -- | The name of the binary cache to push to+    daemonCacheName :: BinaryCacheName,+    -- | The binary cache to push to+    daemonBinaryCache :: BinaryCache,+    -- | The state of active push requests+    daemonPushManager :: PushManagerEnv,+    -- | A multiplexer for push events.+    daemonSubscriptionManager :: SubscriptionManager Protocol.PushRequestId PushEvent,+    -- | Logging env+    daemonLogger :: Logger,+    -- | Shutdown latch+    daemonShutdownLatch :: ShutdownLatch,+    -- | The PID of the daemon process+    daemonPid :: ProcessID+  }++newtype Daemon a = Daemon+  { unDaemon :: ReaderT DaemonEnv IO a+  }+  deriving newtype+    ( Functor,+      Applicative,+      Monad,+      MonadIO,+      MonadReader DaemonEnv,+      MonadUnliftIO,+      MonadCatch,+      MonadMask,+      MonadThrow+    )++instance Katip.Katip Daemon where+  getLogEnv = Log.getKatipLogEnv <$> asks daemonLogger+  localLogEnv f (Daemon m) = Daemon (local (\s -> s {daemonLogger = Log.localLogEnv f (daemonLogger s)}) m)++instance Katip.KatipContext Daemon where+  getKatipContext = Log.getKatipContext <$> asks daemonLogger+  localKatipContext f (Daemon m) = Daemon (local (\s -> s {daemonLogger = Log.localKatipContext f (daemonLogger s)}) m)++  getKatipNamespace = Log.getKatipNamespace <$> asks daemonLogger+  localKatipNamespace f (Daemon m) = Daemon (local (\s -> s {daemonLogger = Log.localKatipNamespace f (daemonLogger s)}) m)++-- | Run a pre-configured daemon.+runDaemon :: DaemonEnv -> Daemon a -> IO a+runDaemon env f = do+  Log.withLogger (daemonLogger env) $ \logger -> do+    let pushManagerEnv = (daemonPushManager env) {pmLogger = logger}+    unDaemon f `runReaderT` env {daemonLogger = logger, daemonPushManager = pushManagerEnv}
+ src/Cachix/Client/Daemon/Types/Log.hs view
@@ -0,0 +1,32 @@+module Cachix.Client.Daemon.Types.Log+  ( LogLevel (..),+    Logger (..),+  )+where++import qualified Katip+import Protolude++-- | The log level to use for logging+--+-- TODO: reuse in deploy agent+data LogLevel+  = Debug+  | Info+  | Warning+  | Error+  deriving stock (Eq, Ord, Show)++data Logger = Logger+  { -- | An optional handle to output logs to.+    -- Defaults to stdout.+    logHandle :: Maybe Handle,+    -- | The log level to use for logging+    logLevel :: LogLevel,+    -- | Logger namespace+    logKNamespace :: Katip.Namespace,+    -- | Logger context+    logKContext :: Katip.LogContexts,+    -- | Logger env+    logKLogEnv :: Katip.LogEnv+  }
+ src/Cachix/Client/Daemon/Types/PushEvent.hs view
@@ -0,0 +1,41 @@+module Cachix.Client.Daemon.Types.PushEvent+  ( PushEvent (..),+    PushEventMessage (..),+    PushRetryStatus (..),+    newPushRetryStatus,+  )+where++import qualified Cachix.Client.Daemon.Protocol as Protocol+import Control.Retry (RetryStatus (..))+import Data.Aeson (FromJSON, ToJSON)+import Data.Time (UTCTime)+import Protolude++data PushEvent = PushEvent+  { eventTimestamp :: UTCTime,+    eventPushId :: Protocol.PushRequestId,+    eventMessage :: PushEventMessage+  }+  deriving stock (Eq, Generic, Show)+  deriving anyclass (FromJSON, ToJSON)++instance Ord PushEvent where+  compare = compare `on` eventTimestamp++data PushEventMessage+  = PushStarted+  | PushStorePathAttempt FilePath Int64 PushRetryStatus+  | PushStorePathProgress FilePath Int64 Int64+  | PushStorePathDone FilePath+  | PushStorePathFailed FilePath Text+  | PushFinished+  deriving stock (Eq, Generic, Show)+  deriving anyclass (FromJSON, ToJSON)++data PushRetryStatus = PushRetryStatus {retryCount :: Int}+  deriving stock (Eq, Generic, Show)+  deriving anyclass (FromJSON, ToJSON)++newPushRetryStatus :: RetryStatus -> PushRetryStatus+newPushRetryStatus RetryStatus {..} = PushRetryStatus {retryCount = rsIterNumber}
+ src/Cachix/Client/Daemon/Types/PushManager.hs view
@@ -0,0 +1,114 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}++module Cachix.Client.Daemon.Types.PushManager+  ( PushManagerEnv (..),+    PushManager (..),+    PushJob (..),+    JobStatus (..),+    JobStats (..),+    PushResult (..),+    OnPushEvent,+    Task (..),+  )+where++import qualified Cachix.Client.Daemon.Log as Log+import qualified Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.Types.Log (Logger)+import Cachix.Client.Daemon.Types.PushEvent (PushEvent (..))+import Control.Concurrent.STM.TBMQueue+import Control.Concurrent.STM.TVar+import Control.Monad.Catch+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Data.HashMap.Strict (HashMap)+import Data.Time (UTCTime)+import qualified Katip+import Protolude++data Task+  = ResolveClosure Protocol.PushRequestId+  | PushStorePath FilePath++type PushJobStore = TVar (HashMap Protocol.PushRequestId PushJob)++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.+    pmStorePathReferences :: TVar (HashMap FilePath [Protocol.PushRequestId]),+    -- | FIFO queue of push tasks.+    pmTaskQueue :: TBMQueue Task,+    pmTaskSemaphore :: QSem,+    -- | Callback for push events.+    pmOnPushEvent :: OnPushEvent,+    pmLogger :: Logger+  }++type OnPushEvent = Protocol.PushRequestId -> PushEvent -> IO ()++newtype PushManager a = PushManager {unPushManager :: ReaderT PushManagerEnv IO a}+  deriving newtype+    ( Functor,+      Applicative,+      Monad,+      MonadIO,+      MonadUnliftIO,+      MonadReader PushManagerEnv,+      MonadCatch,+      MonadMask,+      MonadThrow+    )++instance Katip.Katip PushManager where+  getLogEnv = Log.getKatipLogEnv <$> asks pmLogger+  localLogEnv f (PushManager m) = PushManager (local (\s -> s {pmLogger = Log.localLogEnv f (pmLogger s)}) m)++instance Katip.KatipContext PushManager where+  getKatipContext = Log.getKatipContext <$> asks pmLogger+  localKatipContext f (PushManager m) = PushManager (local (\s -> s {pmLogger = Log.localKatipContext f (pmLogger s)}) m)++  getKatipNamespace = Log.getKatipNamespace <$> asks pmLogger+  localKatipNamespace f (PushManager m) = PushManager (local (\s -> s {pmLogger = Log.localKatipNamespace f (pmLogger s)}) m)++data JobStatus = Queued | Running | Completed | Failed+  deriving stock (Eq, Show)++-- | A push request that has been queued for processing.+data PushJob = PushJob+  { -- | A unique identifier for this push request.+    pushId :: Protocol.PushRequestId,+    -- | The original push request.+    pushRequest :: Protocol.PushRequest,+    -- | The current status of the push job.+    pushStatus :: JobStatus,+    -- | Paths that need to be pushed.+    pushQueue :: Set FilePath,+    -- | Track whether paths were pushed, skipped, or failed to push.+    pushResult :: PushResult,+    -- | Timing stats for the push job.+    pushStats :: JobStats+  }+  deriving stock (Eq, Show)++data PushResult = PushResult+  { prPushedPaths :: Set FilePath,+    prFailedPaths :: Set FilePath,+    prSkippedPaths :: Set FilePath+  }+  deriving stock (Eq, Show)++instance Semigroup PushResult where+  PushResult a1 b1 c1 <> PushResult a2 b2 c2 =+    PushResult (a1 <> a2) (b1 <> b2) (c1 <> c2)++instance Monoid PushResult where+  mempty = PushResult mempty mempty mempty++data JobStats = JobStats+  { jsCreatedAt :: UTCTime,+    jsStartedAt :: Maybe UTCTime,+    jsCompletedAt :: Maybe UTCTime+  }+  deriving stock (Eq, Show)
+ src/Cachix/Client/Daemon/Worker.hs view
@@ -0,0 +1,61 @@+module Cachix.Client.Daemon.Worker+  ( startWorkers,+    stopWorkers,+    startWorker,+    stopWorker,+  )+where++import qualified Control.Concurrent.Async as Async+import Control.Concurrent.STM.TBMQueue+import qualified Control.Immortal as Immortal+import Control.Monad.IO.Unlift (MonadUnliftIO)+import qualified Katip+import Protolude++startWorkers ::+  (MonadUnliftIO m, Katip.KatipContext m) =>+  Int ->+  TBMQueue a ->+  (a -> m ()) ->+  m [Immortal.Thread]+startWorkers numWorkers queue f = do+  replicateM numWorkers (startWorker queue f)++startWorker ::+  (MonadUnliftIO m, Katip.KatipContext m) =>+  TBMQueue a ->+  (a -> m ()) ->+  m Immortal.Thread+startWorker queue f = do+  Immortal.createWithLabel "worker" $ \thread -> do+    Katip.katipAddNamespace "worker" $+      Immortal.onUnexpectedFinish thread logWorkerException (runWorker queue f)++stopWorkers :: (Katip.KatipContext m) => [Immortal.Thread] -> m ()+stopWorkers workers = do+  Katip.logFM Katip.DebugS "Waiting for workers to finish..."+  liftIO $ Async.mapConcurrently_ stopWorker workers+  Katip.logFM Katip.DebugS "Workers finished."++stopWorker :: (MonadIO m) => Immortal.Thread -> m ()+stopWorker worker = liftIO $ do+  Immortal.mortalize worker+  Immortal.wait worker++logWorkerException :: (Exception e, Katip.KatipContext m) => Either e () -> m ()+logWorkerException (Left err) =+  Katip.logFM Katip.ErrorS $ Katip.ls (toS $ displayException err :: Text)+logWorkerException _ = return ()++runWorker :: (MonadIO m) => TBMQueue a -> (a -> m ()) -> m ()+runWorker queue f = loop+  where+    loop = do+      mres <- liftIO $ atomically $ readTBMQueue queue++      case mres of+        Nothing -> return ()+        Just job -> do+          f job+          loop
src/Cachix/Client/Env.hs view
@@ -11,8 +11,6 @@ import qualified Cachix.Client.OptionsParser as Options import Cachix.Client.URI (getBaseUrl) import Cachix.Client.Version (cachixVersion)-import qualified Hercules.CNix as CNix-import qualified Hercules.CNix.Util as CNix.Util import Network.HTTP.Client   ( ManagerSettings,     managerModifyRequest,@@ -25,7 +23,6 @@ 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,@@ -35,17 +32,6 @@  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/Exception.hs view
@@ -18,6 +18,8 @@   | ArtifactNotFound Text   | AccessDeniedBinaryCache Text   | BinaryCacheNotFound Text+  | ImportUnsupportedHash Text+  | RemoveCacheUnsupported Text   deriving (Show, Typeable)  instance Exception CachixException where@@ -36,3 +38,5 @@   displayException (DeprecatedCommand s) = toS s   displayException (AccessDeniedBinaryCache s) = toS s   displayException (BinaryCacheNotFound s) = toS s+  displayException (ImportUnsupportedHash s) = toS s+  displayException (RemoveCacheUnsupported s) = toS s
src/Cachix/Client/InstallationMode.hs view
@@ -163,7 +163,7 @@   where     host = URI.toByteString (URI.appendSubdomain name uri) removeBinaryCache _ _ _ = do-  putText "Removing binary caches is only supported for nix.conf"+  throwIO $ RemoveCacheUnsupported "Removing binary caches is only supported for nix.conf"  nixosBinaryCache :: Config -> BinaryCache.BinaryCache -> UseOptions -> IO () nixosBinaryCache config bc UseOptions {useNixOSFolder = baseDirectory} = do
src/Cachix/Client/OptionsParser.hs view
@@ -6,6 +6,7 @@     DaemonOptions (..),     PushArguments (..),     PushOptions (..),+    defaultPushOptions,     PinOptions (..),     BinaryCacheName,     getOpts,@@ -81,6 +82,7 @@   | Daemon DaemonCommand   | GenerateKeypair BinaryCacheName   | Push PushArguments+  | Import PushOptions Text URI   | Pin PinOptions   | WatchStore PushOptions Text   | WatchExec PushOptions Text Text [Text]@@ -112,10 +114,32 @@   }   deriving (Show) +defaultCompressionLevel :: Int+defaultCompressionLevel = 2++defaultCompressionMethod :: Maybe BinaryCache.CompressionMethod+defaultCompressionMethod = Nothing++defaultNumJobs :: Int+defaultNumJobs = 8++defaultOmitDeriver :: Bool+defaultOmitDeriver = False++defaultPushOptions :: PushOptions+defaultPushOptions =+  PushOptions+    { compressionLevel = defaultCompressionLevel,+      compressionMethod = defaultCompressionMethod,+      numJobs = defaultNumJobs,+      omitDeriver = defaultOmitDeriver+    }+ data DaemonCommand-  = DaemonPushPaths DaemonOptions BinaryCacheName [FilePath]-  | DaemonRun DaemonOptions PushOptions+  = DaemonPushPaths DaemonOptions [FilePath]+  | DaemonRun DaemonOptions PushOptions BinaryCacheName   | DaemonStop DaemonOptions+  | DaemonWatchExec PushOptions BinaryCacheName Text [Text]   deriving (Show)  data DaemonOptions = DaemonOptions@@ -126,22 +150,23 @@ commandParser :: Parser CachixCommand commandParser =   subparser $-    command "authtoken" (infoH authtoken (progDesc "Configure authentication token for communication to HTTP API"))+    command "authtoken" (infoH authtoken (progDesc "Configure an authentication token for Cachix"))       <> command "config" (Config <$> Config.parser)-      <> (hidden <> command "daemon" (infoH (Daemon <$> daemon) (progDesc "Run a daemon that listens push requests over a unix socket")))-      <> command "generate-keypair" (infoH generateKeypair (progDesc "Generate signing key pair for a binary cache"))+      <> (hidden <> command "daemon" (infoH (Daemon <$> daemon) (progDesc "Run a daemon that listens to push requests over a unix socket")))+      <> command "generate-keypair" (infoH generateKeypair (progDesc "Generate a signing key pair for a binary cache"))       <> command "push" (infoH push (progDesc "Upload Nix store paths to a binary cache"))+      <> command "import" (infoH import' (progDesc "Import the contents of a binary cache from an S3-compatible object storage service into Cachix"))       <> command "pin" (infoH pin (progDesc "Pin a store path to prevent it from being garbage collected"))-      <> 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"))+      <> command "watch-exec" (infoH watchExec (progDesc "Run a command while watching /nix/store for newly added store paths and upload them to a binary cache"))+      <> command "watch-store" (infoH watchStore (progDesc "Watch /nix/store for newly added store paths and upload them to a binary cache"))+      <> command "use" (infoH use (progDesc "Configure a binary cache in nix.conf"))       <> command "remove" (infoH remove (progDesc "Remove a binary cache from nix.conf"))-      <> command "deploy" (infoH (DeployCommand <$> DeployOptions.parser) (progDesc "Cachix Deploy commands"))+      <> command "deploy" (infoH (DeployCommand <$> DeployOptions.parser) (progDesc "Manage remote Nix-based systems with Cachix Deploy"))   where     nameArg = strArgument (metavar "CACHE-NAME")     authtoken = AuthToken <$> (stdinFlag <|> (Just <$> authTokenArg))       where-        stdinFlag = flag' Nothing (long "stdin" <> help "Read the token from stdin rather than accepting it as an argument.")+        stdinFlag = flag' Nothing (long "stdin" <> help "Read the auth token from stdin")         authTokenArg = strArgument (metavar "AUTH-TOKEN")     generateKeypair = GenerateKeypair <$> nameArg     validatedLevel l =@@ -162,9 +187,9 @@               <> short 'c'               <> metavar "[0..16]"               <> help-                "The compression level for XZ compression between 0-9 and ZSTD 0-16."+                "The compression level to use. Supported range: [0-9] for xz and [0-16] for zstd."               <> showDefault-              <> value 2+              <> value defaultCompressionLevel           )         <*> option           (eitherReader validatedMethod)@@ -172,22 +197,23 @@               <> short 'm'               <> metavar "xz | zstd"               <> help-                "The compression method, either xz or zstd. Defaults to zstd."-              <> value Nothing+                "The compression method to use. Supported methods: xz | zstd. Defaults to zstd."+              <> value defaultCompressionMethod           )         <*> option           auto           ( long "jobs"               <> short 'j'-              <> help "Number of threads used for pushing store paths."+              <> help "The number of threads to use when pushing store paths."               <> showDefault-              <> value 8+              <> value defaultNumJobs           )         <*> 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)     pushPaths =       (\paths opts cache -> PushPaths opts cache paths)         <$> many (strArgument (metavar "PATHS..."))+    import' = Import <$> pushOptions <*> nameArg <*> strArgument (metavar "S3-URI" <> help "e.g. s3://mybucket?endpoint=https://myexample.com&region=eu-central-1")     keepParser = daysParser <|> revisionsParser <|> foreverParser <|> pure Nothing     -- these three flag are mutually exclusive     daysParser = Just . Days <$> option auto (long "keep-days" <> metavar "INT")@@ -206,9 +232,11 @@         command "push" (infoH daemonPush (progDesc "Push store paths to the daemon"))           <> command "run" (infoH daemonRun (progDesc "Launch the daemon"))           <> command "stop" (infoH daemonStop (progDesc "Stop the daemon and wait for any queued paths to be pushed"))-    daemonPush = DaemonPushPaths <$> daemonOptions <*> nameArg <*> many (strArgument (metavar "PATHS..."))-    daemonRun = DaemonRun <$> daemonOptions <*> pushOptions+          <> command "watch-exec" (infoH daemonWatchExec (progDesc "Run a command and upload any store paths built during its execution"))+    daemonPush = DaemonPushPaths <$> daemonOptions <*> many (strArgument (metavar "PATHS..."))+    daemonRun = DaemonRun <$> daemonOptions <*> pushOptions <*> nameArg     daemonStop = DaemonStop <$> daemonOptions+    daemonWatchExec = DaemonWatchExec <$> pushOptions <*> nameArg <*> strArgument (metavar "CMD") <*> many (strArgument (metavar "-- ARGS"))     daemonOptions = DaemonOptions <$> optional (strOption (long "socket" <> short 's' <> metavar "SOCKET"))     watchExec = WatchExec <$> pushOptions <*> nameArg <*> strArgument (metavar "CMD") <*> many (strArgument (metavar "-- ARGS"))     watchStore = WatchStore <$> pushOptions <*> nameArg@@ -230,7 +258,8 @@                       (maybeReader InstallationMode.fromString)                       ( long "mode"                           <> short 'm'-                          <> help "Mode in which to configure binary caches for Nix. Supported values: nixos, root-nixconf, user-nixconf"+                          <> metavar "nixos | root-nixconf | user-nixconf"+                          <> help "Mode in which to configure binary caches for Nix. Supported values: nixos | root-nixconf | user-nixconf"                       )                   )                 <*> strOption@@ -252,7 +281,8 @@ getOpts :: IO (Flags, CachixCommand) getOpts = do   configpath <- Config.getDefaultFilename-  customExecParser (prefs showHelpOnEmpty) (optsInfo configpath)+  let preferences = showHelpOnError <> showHelpOnEmpty <> helpShowGlobals <> subparserInline+  customExecParser (prefs preferences) (optsInfo configpath)  optsInfo :: Config.ConfigPath -> ParserInfo (Flags, CachixCommand) optsInfo configpath = infoH parser desc
src/Cachix/Client/Push.hs view
@@ -1,22 +1,38 @@ {-# LANGUAGE DataKinds #-}-{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE Rank2Types #-}-{-# LANGUAGE TypeOperators #-}  {- This is a standalone module so it shouldn't depend on any CLI state like Env -} module Cachix.Client.Push   ( -- * Pushing a single path     pushSingleStorePath,     uploadStorePath,++    -- * Push strategy and parameters     PushParams (..),     PushSecret (..),+    getAuthTokenFromPushSecret,     PushStrategy (..),     defaultWithXzipCompressor,     defaultWithXzipCompressorWithLevel,     defaultWithZstdCompressor,     defaultWithZstdCompressorWithLevel,-    findPushSecret, +    -- * Path info+    PathInfo (..),+    newPathInfoFromStorePath,+    newPathInfoFromNarInfo,++    -- * Streaming upload+    Push.S3.UploadResult (..),+    UploadNarDetails (..),+    streamUploadNar,+    streamCopy,+    completeNarUpload,++    -- * Narinfo+    narinfoExists,+    newNarInfoCreate,+     -- * Pushing a closure of store paths     pushClosure,     getMissingPathsForClosure,@@ -27,10 +43,9 @@ import qualified Cachix.API as API import Cachix.API.Error 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.Retry (retryAll, retryHttp) import Cachix.Client.Secrets import Cachix.Client.Servant import qualified Cachix.Types.BinaryCache as BinaryCache@@ -40,7 +55,7 @@ import Conduit (MonadUnliftIO) import Control.Concurrent.Async (mapConcurrently) import qualified Control.Concurrent.QSem as QSem-import Control.Exception.Safe (MonadMask, throwM)+import Control.Exception.Safe (MonadCatch, MonadMask, throwM) import Control.Monad.Trans.Resource (ResourceT) import Control.Retry (RetryStatus) import Crypto.Sign.Ed25519@@ -50,13 +65,13 @@ import qualified Data.Conduit.Zstd as Zstd (compress) import Data.IORef import qualified Data.Set as Set-import Data.String.Here import qualified Data.Text as T import Hercules.CNix (StorePath) import qualified Hercules.CNix.Std.Set as Std.Set import Hercules.CNix.Store (Store) import qualified Hercules.CNix.Store as Store import Network.HTTP.Types (status401, status404)+import qualified Nix.NarInfo as NarInfo import Protolude hiding (toS) import Protolude.Conv import Servant.API@@ -64,14 +79,24 @@ import Servant.Auth.Client import Servant.Client.Streaming import Servant.Conduit ()-import System.Environment (lookupEnv) import qualified System.Nix.Base32 import System.Nix.Nar +-- | A secret for authenticating with a cache. data PushSecret-  = PushToken Token-  | PushSigningKey Token SigningKey+  = -- | An auth token. Could be a personal, cache, or agent token.+    PushToken Token+  | -- | An auth token with a signing key for pushing to self-signed caches.+    PushSigningKey Token SigningKey +getAuthTokenFromPushSecret :: PushSecret -> Maybe Token+getAuthTokenFromPushSecret = \case+  PushToken token -> nullTokenToMaybe token+  PushSigningKey token _ -> nullTokenToMaybe token+  where+    nullTokenToMaybe (Token "") = Nothing+    nullTokenToMaybe (Token token) = Just $ Token token+ -- | Parameters for pushing a closure of store paths, to be passed to 'pushClosure'. -- This also contains the parameters for pushing a single path, in 'pushParamStrategy'. data PushParams m r = PushParams@@ -134,123 +159,66 @@   StorePath ->   -- | r is determined by the 'PushStrategy'   m r-pushSingleStorePath cache storePath = retryAll $ \retrystatus -> do+pushSingleStorePath pushParams storePath = retryAll $ \retrystatus -> do   storeHash <- liftIO $ Store.getStorePathHash storePath-  let name = pushParamsName cache-      strategy = pushParamsStrategy cache storePath-  -- Check if narinfo already exists-  res <--    liftIO $-      (`runClientM` pushParamsClientEnv cache) $-        API.narinfoHead-          cachixClient-          (getCacheAuthToken (pushParamsSecret cache))-          name-          (NarInfoHash.NarInfoHash (decodeUtf8With lenientDecode storeHash))+  let strategy = pushParamsStrategy pushParams storePath+  res <- liftIO $ narinfoExists pushParams storeHash   case res of     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 status404 -> uploadStorePath pushParams storePath retrystatus       | isErr err status401 -> on401 strategy err       | otherwise -> onError strategy err +narinfoExists :: PushParams m r -> ByteString -> IO (Either ClientError NoContent)+narinfoExists pushParams storeHash = do+  let cacheName = pushParamsName pushParams+      authToken = getCacheAuthToken (pushParamsSecret pushParams)+  retryHttp $+    (`runClientM` pushParamsClientEnv pushParams) $+      API.narinfoHead+        cachixClient+        authToken+        cacheName+        (NarInfoHash.NarInfoHash (decodeUtf8With lenientDecode storeHash))+ getCacheAuthToken :: PushSecret -> Token getCacheAuthToken (PushToken token) = token getCacheAuthToken (PushSigningKey token _) = token  uploadStorePath ::-  (MonadUnliftIO m) =>+  (MonadUnliftIO m, MonadCatch m) =>   -- | details for pushing to cache   PushParams m r ->   StorePath ->   RetryStatus ->   -- | r is determined by the 'PushStrategy'   m r-uploadStorePath cache storePath retrystatus = do-  let store = pushParamsStore cache+uploadStorePath pushParams storePath retrystatus = do+  let store = pushParamsStore pushParams+      strategy = pushParamsStrategy pushParams storePath+   -- TODO: storePathText is redundant. Use storePath directly.   storePathText <- liftIO $ Store.storePathToPath store storePath-  let (storeHash, storeSuffix) = splitStorePath $ toS storePathText-      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-  let storePathSize = Store.validPathInfoNarSize pathinfo-  onAttempt strategy retrystatus storePathSize--  withCompressor $ \compressor -> do-    uploadResult <--      runConduitRes $-        streamNarIO narEffectsIO (toS storePathText) Data.Conduit.yield-          .| passthroughSizeSink narSizeRef-          .| passthroughHashSink narHashRef-          .| onUncompressedNARStream strategy retrystatus storePathSize-          .| compressor-          .| passthroughSizeSink fileSizeRef-          .| passthroughHashSinkB16 fileHashRef-          .| Push.S3.streamUpload cacheClientEnv authToken cacheName (compressionMethod strategy)+  normalized <- liftIO $ Store.followLinksToStorePath store storePathText+  pathInfo <- newPathInfoFromStorePath store normalized+  let narSize = fromIntegral (piNarSize pathInfo) -    case uploadResult of-      Left err -> throwIO err-      Right (narId, uploadId, parts) -> 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 = 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+  onAttempt strategy retrystatus narSize -        -- 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+  eresult <-+    runConduitRes $+      streamNarIO narEffectsIO (toS storePathText) Data.Conduit.yield+        .| streamUploadNar pushParams storePath narSize retrystatus -  onDone strategy+  case eresult of+    Left e -> onError strategy e+    Right (uploadResult, uploadNarDetails) -> do+      nic <- newNarInfoCreate pushParams storePath pathInfo uploadNarDetails+      completeNarUpload pushParams uploadResult nic+      onDone strategy  -- | Push an entire closure --@@ -269,9 +237,227 @@ pushClosure traversal pushParams inputStorePaths = do   (allPaths, missingPaths) <- getMissingPathsForClosure pushParams inputStorePaths   paths <- pushOnClosureAttempt pushParams allPaths missingPaths-  traversal (\storePath -> retryAll $ \retrystatus -> uploadStorePath pushParams storePath retrystatus) paths+  flip traversal paths $ \storePath ->+    retryAll $ uploadStorePath pushParams storePath -getMissingPathsForClosure :: (MonadIO m, MonadMask m) => PushParams m r -> [StorePath] -> m ([StorePath], [StorePath])+completeNarUpload ::+  (MonadUnliftIO m) =>+  PushParams n r ->+  Push.S3.UploadResult ->+  Api.NarInfoCreate ->+  m ()+completeNarUpload pushParams Push.S3.UploadResult {..} nic = do+  let cacheName = pushParamsName pushParams+      authToken = getCacheAuthToken (pushParamsSecret pushParams)+      clientEnv = pushParamsClientEnv pushParams++  -- Complete the multipart upload and upload the narinfo+  let completeMultipartUploadRequest =+        API.completeNarUpload cachixClient authToken cacheName urNarId urUploadId $+          Multipart.CompletedMultipartUpload+            { Multipart.parts = urParts,+              Multipart.narInfoCreate = nic+            }++  liftIO $ void $ retryHttp $ withClientM completeMultipartUploadRequest clientEnv escalate++data UploadNarDetails = UploadNarDetails+  { undNarSize :: Integer,+    undNarHash :: Text,+    undFileSize :: Integer,+    undFileHash :: Text+  }+  deriving stock (Eq, Show)++-- | A simplified type for ValidPathInfo.+-- Can be constructed either from an existing NarInfo, or by querying the Nix store.+-- Only includes the common fields between remote and local path infos.+data PathInfo = PathInfo+  { piPath :: FilePath,+    piNarHash :: Text,+    piNarSize :: Integer,+    piDeriver :: Maybe Text,+    piReferences :: Set FilePath+  }+  deriving stock (Eq, Show)++-- | Create a 'PathInfo' for a store path by querying the Nix store.+newPathInfoFromStorePath :: (MonadIO m) => Store -> StorePath -> m PathInfo+newPathInfoFromStorePath store storePath = liftIO $ do+  pathInfo <- Store.queryPathInfo store storePath++  path <- Store.storePathToPath store storePath+  narHash <- Store.validPathInfoNarHash32 pathInfo+  let narSize = fromIntegral $ Store.validPathInfoNarSize pathInfo+  deriver <-+    Store.validPathInfoDeriver store pathInfo+      >>= mapM Store.getStorePathBaseName+  references <-+    Store.validPathInfoReferences store pathInfo+      >>= mapM Store.getStorePathBaseName++  return $+    PathInfo+      { piPath = toS path,+        piNarHash = decodeUtf8 narHash,+        piNarSize = narSize,+        piDeriver = fmap toS deriver,+        piReferences = Set.fromList (fmap toS references)+      }++-- | Create a 'PathInfo' for a store path from an existing NarInfo.+newPathInfoFromNarInfo :: (Applicative m) => NarInfo.SimpleNarInfo -> m PathInfo+newPathInfoFromNarInfo narInfo =+  pure $+    PathInfo+      { piPath = NarInfo.storePath narInfo,+        piNarHash = NarInfo.narHash narInfo,+        piNarSize = NarInfo.narSize narInfo,+        piDeriver = NarInfo.deriver narInfo,+        piReferences = NarInfo.references narInfo+      }++-- | A conduit that compresses and streams a NAR to a cache.+streamUploadNar ::+  (MonadUnliftIO m) =>+  PushParams m r ->+  StorePath ->+  Int64 ->+  RetryStatus ->+  ConduitT ByteString Void (ResourceT m) (Either ClientError (Push.S3.UploadResult, UploadNarDetails))+streamUploadNar pushParams storePath storePathSize retrystatus = do+  let cacheName = pushParamsName pushParams+      authToken = getCacheAuthToken (pushParamsSecret pushParams)+      clientEnv = pushParamsClientEnv pushParams++  let strategy = pushParamsStrategy pushParams storePath+      withCompressor = case compressionMethod strategy of+        BinaryCache.XZ -> defaultWithXzipCompressorWithLevel (compressionLevel strategy)+        BinaryCache.ZSTD -> defaultWithZstdCompressorWithLevel (compressionLevel strategy)++  narSizeRef <- liftIO $ newIORef 0+  fileSizeRef <- liftIO $ newIORef 0+  narHashRef <- liftIO $ newIORef ("" :: ByteString)+  fileHashRef <- liftIO $ newIORef ("" :: ByteString)++  result <- withCompressor $ \compressor ->+    awaitForever Data.Conduit.yield+      .| passthroughSizeSink narSizeRef+      .| passthroughHashSink narHashRef+      .| onUncompressedNARStream strategy retrystatus storePathSize+      .| compressor+      .| passthroughSizeSink fileSizeRef+      .| passthroughHashSinkB16 fileHashRef+      .| Push.S3.streamUpload clientEnv authToken cacheName (compressionMethod strategy)++  for result $ \uploadResult -> liftIO $ do+    narSize <- readIORef narSizeRef+    narHash <- ("sha256:" <>) . System.Nix.Base32.encode <$> readIORef narHashRef+    fileSize <- readIORef fileSizeRef+    fileHash <- readIORef fileHashRef++    let uploadNarDetails =+          UploadNarDetails+            { undNarSize = narSize,+              undNarHash = narHash,+              undFileSize = fileSize,+              undFileHash = toS fileHash+            }++    return (uploadResult, uploadNarDetails)++-- | A conduit that streams an existing NAR to a cache.+-- Used to copy NARs between caches, e.g. S3 -> Cachix.+streamCopy ::+  (MonadUnliftIO m) =>+  PushParams m r ->+  StorePath ->+  Int64 ->+  RetryStatus ->+  BinaryCache.CompressionMethod ->+  ConduitT ByteString Void (ResourceT m) (Either ClientError (Push.S3.UploadResult, UploadNarDetails))+streamCopy pushParams storePath claimedFileSize retrystatus compressionMethod = do+  let cacheName = pushParamsName pushParams+      authToken = getCacheAuthToken (pushParamsSecret pushParams)+      clientEnv = pushParamsClientEnv pushParams+      strategy = pushParamsStrategy pushParams storePath++  fileSizeRef <- liftIO $ newIORef 0+  fileHashRef <- liftIO $ newIORef ("" :: ByteString)++  result <-+    awaitForever Data.Conduit.yield+      .| onUncompressedNARStream strategy retrystatus claimedFileSize+      .| passthroughSizeSink fileSizeRef+      .| passthroughHashSinkB16 fileHashRef+      .| Push.S3.streamUpload clientEnv authToken cacheName compressionMethod++  for result $ \uploadResult -> liftIO $ do+    fileSize <- readIORef fileSizeRef+    fileHash <- readIORef fileHashRef++    let uploadNarDetails =+          UploadNarDetails+            { undNarSize = 0,+              undNarHash = "",+              undFileSize = fileSize,+              undFileHash = toS fileHash+            }++    return (uploadResult, uploadNarDetails)++newNarInfoCreate ::+  (MonadIO m, MonadCatch m) =>+  PushParams n r ->+  StorePath ->+  PathInfo ->+  UploadNarDetails ->+  m Api.NarInfoCreate+newNarInfoCreate pushParams storePath pathInfo UploadNarDetails {..} = do+  let store = pushParamsStore pushParams+  let strategy = pushParamsStrategy pushParams storePath+  storeDir <- Store.storeDir store+  storePathText <- liftIO $ toS <$> Store.storePathToPath store storePath++  -- TODO: show expected vs actual NAR hash+  when (undNarHash /= piNarHash pathInfo) $+    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"++  let deriver =+        fromMaybe "unknown-deriver" $+          if omitDeriver strategy+            then Nothing+            else piDeriver pathInfo++  let references = fmap toS $ Set.toList $ piReferences pathInfo+  let fpReferences = fmap (\fp -> toS storeDir <> "/" <> fp) references++  let (storeHash, storeSuffix) = splitStorePath storePathText+  let fp = fingerprint storePathText undNarHash undNarSize fpReferences+      sig = case pushParamsSecret pushParams of+        PushToken _ -> Nothing+        PushSigningKey _ signKey -> Just $ toS $ B64.encode $ unSignature $ dsign (signingSecretKey signKey) fp++  let nic =+        Api.NarInfoCreate+          { Api.cStoreHash = storeHash,+            Api.cStoreSuffix = storeSuffix,+            Api.cNarHash = undNarHash,+            Api.cNarSize = undNarSize,+            Api.cFileSize = undFileSize,+            Api.cFileHash = undFileHash,+            Api.cReferences = references,+            Api.cDeriver = deriver,+            Api.cSig = sig+          }++  escalate $ Api.isNarInfoCreateValid nic++  return nic++getMissingPathsForClosure :: (MonadIO m, MonadMask m) => PushParams n r -> [StorePath] -> m ([StorePath], [StorePath]) getMissingPathsForClosure pushParams inputPaths = do   let store = pushParamsStore pushParams       clientEnv = pushParamsClientEnv pushParams@@ -284,7 +470,7 @@   hashes <- liftIO $ for paths (fmap (decodeUtf8With lenientDecode) . Store.getStorePathHash)   -- Check what store paths are missing   missingHashesList <--    retryAll $ \_ ->+    retryHttp $       liftIO $         escalate           =<< API.narinfoBulk@@ -292,7 +478,7 @@             (getCacheAuthToken (pushParamsSecret pushParams))             (pushParamsName pushParams)             hashes-            `runClientM` clientEnv+          `runClientM` clientEnv   let missingHashes = Set.fromList (encodeUtf8 <$> missingHashesList)   pathsAndHashes <- liftIO $     for paths $ \path -> do@@ -300,40 +486,6 @@       pure (hash_, path)   let missing = map snd $ filter (\(hash_, _path) -> Set.member hash_ missingHashes) pathsAndHashes   return (paths, missing)---- TODO: move to a separate module specific to cli---- | Find auth token or signing key in the 'Config' or environment variable-findPushSecret ::-  Config.Config ->-  -- | Cache name-  Text ->-  -- | Secret key or exception-  IO PushSecret-findPushSecret config name = do-  maybeSigningKeyEnv <- toS <<$>> lookupEnv "CACHIX_SIGNING_KEY"-  maybeAuthToken <- Config.getAuthTokenMaybe config-  let maybeSigningKeyConfig = Config.secretKey <$> head (getBinaryCache config)-  case maybeSigningKeyEnv <|> maybeSigningKeyConfig of-    Just signingKey -> escalateAs FatalError $ PushSigningKey (fromMaybe (Token "") maybeAuthToken) <$> parseSigningKeyLenient signingKey-    Nothing -> case maybeAuthToken of-      Just authToken -> return $ PushToken authToken-      Nothing -> throwIO $ NoSigningKey msg-  where-    -- we reverse list of caches to prioritize keys added as last-    getBinaryCache c =-      reverse $-        filter (\bc -> Config.name bc == name) (Config.binaryCaches c)-    msg :: Text-    msg =-      [iTrim|-Neither auth token nor signing key are present.--They are looked up via $CACHIX_AUTH_TOKEN and $CACHIX_SIGNING_KEY,-and if missing also looked up from ~/.config/cachix/cachix.dhall--Read https://mycache.cachix.org for instructions how to push to your binary cache.-    |]  mapConcurrentlyBounded :: (Traversable t) => Int -> (a -> IO b) -> t a -> IO (t b) mapConcurrentlyBounded bound action items = do
src/Cachix/Client/Push/S3.hs view
@@ -6,7 +6,7 @@  import qualified Cachix.API as API import Cachix.API.Error-import Cachix.Client.Retry (retryAll)+import Cachix.Client.Retry (retryHttp) import Cachix.Client.Servant (cachixClient) import Cachix.Types.BinaryCache import qualified Cachix.Types.MultipartUpload as Multipart@@ -52,6 +52,13 @@ outputBufferSize :: Int outputBufferSize = 100 +data UploadResult = UploadResult+  { urNarId :: UUID,+    urUploadId :: Text,+    urParts :: Maybe (NonEmpty Multipart.CompletedPart)+  }+  deriving stock (Eq, Show)+ streamUpload ::   forall m.   (MonadUnliftIO m, MonadResource m) =>@@ -63,30 +70,31 @@     ByteString     Void     m-    (Either SomeException (UUID, Text, Maybe (NonEmpty Multipart.CompletedPart)))+    (Either ClientError UploadResult) 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+  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)+          .| 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)+    createMultipartUpload :: ConduitT ByteString Void m (Either ClientError Multipart.CreateMultipartUploadResponse)+    createMultipartUpload = do+      let createNarRequest = API.createNar cachixClient authToken cacheName (Just compressionMethod)+      liftIO $ retryHttp $ runClientM createNarRequest env      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+      let uploadNarPartRequest =+            API.uploadNarPart cachixClient authToken cacheName narId uploadId partNumber (Multipart.SigningData (decodeUtf8 contentMD5))+      Multipart.UploadPartResponse {uploadUrl} <- liftIO $ retryHttp $ withClientM uploadNarPartRequest env escalate        initialRequest <- liftIO $ HTTP.parseUrlThrow (toS uploadUrl)       let request =@@ -99,7 +107,7 @@                   ]               } -      response <- liftIO $ retryAll $ \_ -> HTTP.httpNoBody request manager+      response <- liftIO $ retryHttp $ HTTP.httpNoBody request manager       let eTag = decodeUtf8 <$> lookup HTTP.hETag (HTTP.responseHeaders response)       -- Strictly evaluate each eTag after uploading each part       let !_ = rwhnf eTag@@ -107,9 +115,15 @@      completeMultipartUpload narId uploadId = do       parts <- CC.sinkList-      return $ Right (narId, uploadId, sequenceA $ NonEmpty.fromList parts)+      return $+        Right $+          UploadResult+            { urNarId = narId,+              urUploadId = uploadId,+              urParts = sequenceA (NonEmpty.fromList parts)+            }      abortMultipartUpload narId uploadId err = do       let abortMultipartUploadRequest = API.abortMultipartUpload cachixClient authToken cacheName narId uploadId-      _ <- liftIO $ withClientM abortMultipartUploadRequest env escalate+      _ <- liftIO $ retryHttp $ withClientM abortMultipartUploadRequest env escalate       return $ Left err
src/Cachix/Client/Retry.hs view
@@ -1,28 +1,38 @@ module Cachix.Client.Retry   ( retryAll,+    retryAllWithPolicy,     retryAllWithLogging,+    retryHttp,+    retryHttpWith,     endlessRetryPolicy,     endlessConstantRetryPolicy,   ) where -import Control.Exception.Safe (Handler (..), MonadMask, isSyncException)-import Control.Retry (RetryPolicy, RetryPolicyM, RetryStatus, constantDelay, exponentialBackoff, limitRetries, logRetries, recoverAll, recovering, skipAsyncExceptions)-import Protolude hiding (Handler (..))--retryAll :: (MonadIO m, MonadMask m) => (RetryStatus -> m a) -> m a-retryAll =-  recoverAll defaultRetryPolicy---- Catches all exceptions except async exceptions with logging support-retryAllWithLogging :: (MonadIO m, MonadMask m) => RetryPolicyM m -> (Bool -> SomeException -> RetryStatus -> m ()) -> m a -> m a-retryAllWithLogging retryPolicy logger action = recovering retryPolicy handlers $ const action-  where-    handlers = skipAsyncExceptions ++ [exitSuccessHandler, loggingHandler]-    exitSuccessHandler :: (MonadIO m) => RetryStatus -> Handler m Bool-    exitSuccessHandler _ = Handler $ \(_ :: ExitCode) -> return False-    loggingHandler = logRetries exceptionPredicate logger-    exceptionPredicate = return . isSyncException+import Cachix.Client.Exception (CachixException (..))+import qualified Control.Concurrent.Async as Async+import Control.Exception.Safe+  ( Handler (..),+    isSyncException,+  )+import Control.Monad.Catch (MonadCatch, MonadMask, handleJust, throwM)+import Control.Retry+import Data.List (lookup)+import Data.Time+  ( UTCTime,+    defaultTimeLocale,+    diffUTCTime,+    getCurrentTime,+    nominalDiffTimeToSeconds,+    readPTime,+    rfc822DateFormat,+  )+import GHC.Read (Read (readPrec))+import qualified Network.HTTP.Client as HTTP+import Network.HTTP.Types.Header (hRetryAfter)+import qualified Network.HTTP.Types.Status as HTTP+import Protolude hiding (Handler (..), handleJust)+import qualified Text.ParserCombinators.ReadPrec as ReadPrec (lift)  defaultRetryPolicy :: RetryPolicy defaultRetryPolicy =@@ -35,3 +45,135 @@ endlessRetryPolicy :: RetryPolicy endlessRetryPolicy =   exponentialBackoff (1000 * 1000)++retryAll :: (MonadIO m, MonadMask m) => (RetryStatus -> m a) -> m a+retryAll = retryAllWithPolicy defaultRetryPolicy++retryAllWithPolicy :: (MonadIO m, MonadMask m) => RetryPolicyM m -> (RetryStatus -> m a) -> m a+retryAllWithPolicy policy f =+  recovering policy handlers $+    rethrowLinkedThreadExceptions . f+  where+    handlers = skipAsyncExceptions ++ [exitCodeHandler, cachixExceptionsHandler, allHandler]++    -- Skip over exitSuccess/exitFailure+    exitCodeHandler _ = Handler $ \(_ :: ExitCode) -> return False++    -- Skip over fatal Cachix exceptions+    cachixExceptionsHandler _ = Handler $ \(_ :: CachixException) -> return False++    -- Retry everything else+    allHandler _ = Handler $ \(_ :: SomeException) -> return True++-- Catches all exceptions except async exceptions with logging support+retryAllWithLogging ::+  (MonadIO m, MonadMask m) =>+  RetryPolicyM m ->+  (Bool -> SomeException -> RetryStatus -> m ()) ->+  m a ->+  m a+retryAllWithLogging policy logger f =+  recovering policy handlers $+    const (rethrowLinkedThreadExceptions f)+  where+    handlers = skipAsyncExceptions ++ [exitCodeHandler, loggingHandler]++    -- Skip over exitSuccess/exitFailure+    exitCodeHandler _ = Handler $ \(_ :: ExitCode) -> return False++    -- Log and retry everything else+    loggingHandler = logRetries (return . isSyncException) logger++-- | Unwrap 'Async.ExceptionInLinkedThread' exceptions and rethrow the inner exception.+rethrowLinkedThreadExceptions :: (MonadCatch m) => m a -> m a+rethrowLinkedThreadExceptions =+  handleJust unwrapLinkedThreadException throwM++unwrapLinkedThreadException :: SomeException -> Maybe SomeException+unwrapLinkedThreadException e+  | Just (Async.ExceptionInLinkedThread _ e') <- fromException e = Just e'+  | otherwise = Nothing++retryHttp :: (MonadIO m, MonadMask m) => m a -> m a+retryHttp = retryHttpWith defaultRetryPolicy++-- | Retry policy for HTTP requests.+--+-- Retries a subset of HTTP exceptions and overrides the delay with the Retry-After header if present.+retryHttpWith :: forall m a. (MonadIO m, MonadMask m) => RetryPolicyM m -> m a -> m a+retryHttpWith policy = recoveringDynamic policy handlers . const+  where+    handlers :: [RetryStatus -> Handler m RetryAction]+    handlers =+      skipAsyncExceptions' ++ [retryHttpExceptions, retrySyncExceptions]++    skipAsyncExceptions' = map (fmap toRetryAction .) skipAsyncExceptions++    retryHttpExceptions _ = Handler httpExceptionToRetryAction+    retrySyncExceptions _ = Handler $ \(_ :: SomeException) -> return ConsultPolicy++    httpExceptionToRetryAction :: HTTP.HttpException -> m RetryAction+    httpExceptionToRetryAction (HTTP.HttpExceptionRequest _ (HTTP.StatusCodeException response _))+      | statusMayHaveRetryHeader (HTTP.responseStatus response) = overrideDelayWithRetryAfter response+    httpExceptionToRetryAction ex = return . toRetryAction . shouldRetryHttpException $ ex++    statusMayHaveRetryHeader :: HTTP.Status -> Bool+    statusMayHaveRetryHeader = flip elem [HTTP.tooManyRequests429, HTTP.serviceUnavailable503]++data RetryAfter+  = RetryAfterDate UTCTime+  | RetryAfterSeconds Int+  deriving (Eq, Show)++instance Read RetryAfter where+  readPrec = parseSeconds <|> parseWebDate+    where+      parseSeconds = RetryAfterSeconds <$> readPrec+      parseWebDate = ReadPrec.lift $ RetryAfterDate <$> readPTime True defaultTimeLocale rfc822DateFormat++overrideDelayWithRetryAfter :: (MonadIO m) => HTTP.Response a -> m RetryAction+overrideDelayWithRetryAfter response =+  case lookupRetryAfter response of+    Nothing ->+      return ConsultPolicy+    Just (RetryAfterSeconds seconds) ->+      return $ ConsultPolicyOverrideDelay (seconds * 1000 * 1000)+    Just (RetryAfterDate date) -> do+      seconds <- secondsFromNow date+      return $+        if seconds > 0+          then ConsultPolicyOverrideDelay (seconds * 1000 * 1000)+          else ConsultPolicy+  where+    secondsFromNow date = do+      now <- liftIO getCurrentTime+      return $ ceiling $ nominalDiffTimeToSeconds (date `diffUTCTime` now)++    lookupRetryAfter = readMaybe . decodeUtf8 <=< lookup hRetryAfter . HTTP.responseHeaders++-- | Determine whether the HTTP exception is worth retrying.+--+-- Temporary connection or network transfer errors are good candidates.+shouldRetryHttpException :: HTTP.HttpException -> Bool+shouldRetryHttpException (HTTP.InvalidUrlException _ _) = False+shouldRetryHttpException (HTTP.HttpExceptionRequest _ reason) =+  case reason of+    HTTP.ConnectionClosed -> True+    HTTP.ConnectionFailure _ -> True+    HTTP.ConnectionTimeout -> True+    HTTP.IncompleteHeaders -> True+    HTTP.InternalException _ -> True+    HTTP.InvalidChunkHeaders -> True+    HTTP.InvalidProxyEnvironmentVariable _ _ -> True+    HTTP.InvalidStatusLine _ -> True+    HTTP.NoResponseDataReceived -> True+    HTTP.ProxyConnectException _ _ status+      | HTTP.statusIsServerError status -> True+    HTTP.ResponseBodyTooShort _ _ -> True+    HTTP.ResponseTimeout -> True+    HTTP.StatusCodeException response _+      | HTTP.responseStatus response == HTTP.tooManyRequests429 -> True+    HTTP.StatusCodeException response _+      | HTTP.statusIsServerError (HTTP.responseStatus response) -> True+    HTTP.HttpZlibException _ -> True+    _ -> False
src/Cachix/Client/URI.hs view
@@ -10,6 +10,7 @@     appendSubdomain,     getPortFor,     getPath,+    getQueryParam,     requiresSSL,     parseURI,     serialize,@@ -81,6 +82,11 @@  getPath :: URI -> ByteString getPath = UBS.uriPath . getUri++getQueryParam :: URI -> ByteString -> Maybe ByteString+getQueryParam uri param =+  let query = UBS.uriQuery $ getUri uri+   in map snd $ head $ filter (\(key, _) -> key == param) $ UBS.queryPairs query  requiresSSL :: UBS.Scheme -> Bool requiresSSL (UBS.Scheme "https") = True
src/Cachix/Deploy/ActivateCommand.hs view
@@ -113,7 +113,7 @@   where     loop = do       deployment <--        Retry.retryAll . const $+        Retry.retryHttp $           escalate <=< (`runClientM` clientEnv) $             API.V1.getDeployment deployClientV1 token deploymentID
src/Cachix/Deploy/Agent.hs view
@@ -17,7 +17,7 @@ import qualified Control.Concurrent.Async as Async import Control.Concurrent.Extra (once) import qualified Control.Concurrent.MVar as MVar-import Control.Exception.Safe (onException, throwString)+import Control.Exception.Safe (onException) import qualified Control.Exception.Safe as Safe import qualified Control.Retry as Retry import qualified Data.Aeson as Aeson@@ -25,10 +25,11 @@ import Data.String (String) import qualified Katip as K import Paths_cachix (getBinDir)-import Protolude hiding (onException, toS)+import Protolude hiding (onException, toS, (<.>)) import Protolude.Conv import qualified System.Directory as Directory import System.Environment (getEnv, lookupEnv)+import System.FilePath ((<.>), (</>)) import qualified System.Posix.Files as Posix.Files import qualified System.Posix.Process as Posix import qualified System.Posix.Signals as Signals@@ -42,72 +43,92 @@   { name :: Text,     token :: Text,     profileName :: Text,-    agentState :: IORef (Maybe WSS.AgentInformation),-    pid :: Posix.CPid,     bootstrap :: Bool,     host :: URI,+    websocket :: ServiceWebSocket,+    agentState :: IORef (Maybe WSS.AgentInformation),     logOptions :: Log.Options,     withLog :: Log.WithLog,-    websocket :: ServiceWebSocket+    lockFile :: FilePath,+    pid :: Posix.CPid,+    pidFile :: FilePath   } +mkAgent :: Log.WithLog -> Log.Options -> Maybe FilePath -> Config.CachixOptions -> CLI.AgentOptions -> Text -> IO Agent+mkAgent withLog logOptions mlockDirectory cachixOptions agentOptions agentToken = do+  agentState <- newIORef Nothing+  pid <- Posix.getProcessID+  lockDirectory <- maybe Lock.getLockDirectory return mlockDirectory++  let host = Config.host cachixOptions+      basename = URI.getHostname host++      agentName = CLI.name agentOptions+      profileName = fromMaybe "system" (CLI.profile agentOptions)++      port =+        host+          & URI.getScheme+          & URI.getPortFor+          & fromMaybe (URI.Port 80)++      websocketOptions =+        WebSocket.Options+          { WebSocket.host = basename,+            WebSocket.port = port,+            WebSocket.path = "/ws",+            WebSocket.useSSL = URI.requiresSSL (URI.getScheme host),+            WebSocket.headers = WebSocket.createHeaders agentName agentToken,+            WebSocket.identifier = agentIdentifier agentName+          }++  websocket <- WebSocket.new withLog websocketOptions++  let lockFilename = "agent-" <> toS agentName++  return $+    Agent+      { name = agentName,+        token = agentToken,+        profileName = profileName,+        bootstrap = CLI.bootstrap agentOptions,+        agentState = agentState,+        pid = pid,+        pidFile = lockDirectory </> lockFilename <.> Lock.pidExtension,+        lockFile = lockDirectory </> lockFilename <.> Lock.lockExtension,+        host = host,+        logOptions = logOptions,+        withLog = withLog,+        websocket = websocket+      }+ agentIdentifier :: Text -> Text agentIdentifier agentName = unwords [agentName, toS versionNumber]  run :: Config.CachixOptions -> CLI.AgentOptions -> IO () run cachixOptions agentOptions =   Log.withLog logOptions $ \withLog ->-    logExceptions withLog $-      withAgentLock agentOptions $ do-        checkUserOwnsHome--        -- TODO: show a more helpful error if the token is missing-        -- TODO: show a more helpful error when the token isn't valid-        -- TODO: wrap the token in a newtype or use servant's Token-        agentToken <- toS <$> getEnv "CACHIX_AGENT_TOKEN"-        agentState <- newIORef Nothing+    logExceptions withLog $ do+      checkUserOwnsHome -        pid <- Posix.getProcessID+      -- TODO: show a more helpful error if the token is missing+      -- TODO: show a more helpful error when the token isn't valid+      -- TODO: wrap the token in a newtype or use servant's Token+      agentToken <- toS <$> getEnv "CACHIX_AGENT_TOKEN" -        let port = fromMaybe (URI.Port 80) $ (URI.getPortFor . URI.getScheme) host-        let websocketOptions =-              WebSocket.Options-                { WebSocket.host = basename,-                  WebSocket.port = port,-                  WebSocket.path = "/ws",-                  WebSocket.useSSL = URI.requiresSSL (URI.getScheme host),-                  WebSocket.headers = WebSocket.createHeaders agentName agentToken,-                  WebSocket.identifier = agentIdentifier agentName-                }+      agent <- mkAgent withLog logOptions Nothing cachixOptions agentOptions agentToken -        websocket <- WebSocket.new withLog websocketOptions-        channel <- WebSocket.receive websocket-        shutdownWebsocket <- connectToService websocket+      withAgentLock agent $ do+        -- Connect to the backend+        channel <- WebSocket.receive (websocket agent)+        shutdownWebsocket <- connectToService (websocket agent) +        -- Shutdown the socket on sigint         installSignalHandlers shutdownWebsocket -        let agent =-              Agent-                { name = agentName,-                  token = agentToken,-                  profileName = profileName,-                  agentState = agentState,-                  pid = pid,-                  bootstrap = CLI.bootstrap agentOptions,-                  host = host,-                  logOptions = logOptions,-                  withLog = withLog,-                  websocket = websocket-                }-         WebSocket.readDataMessages channel $ \message ->           handleCommand agent (WSS.command message)   where-    host = Config.host cachixOptions-    basename = URI.getHostname host-    agentName = CLI.name agentOptions-    profileName = fromMaybe "system" (CLI.profile agentOptions)-     verbosity =       if Config.verbose cachixOptions         then Log.Verbose@@ -120,6 +141,12 @@           environment = "production"         } +handleCommand :: Agent -> WSS.BackendCommand -> IO ()+handleCommand agent command =+  case command of+    WSS.AgentRegistered agentInformation -> registerAgent agent agentInformation+    WSS.Deployment deploymentDetails -> launchDeployment agent deploymentDetails+ logExceptions :: Log.WithLog -> IO () -> IO () logExceptions withLog action =   action `catches` [agentHandler, exceptionHandler]@@ -139,20 +166,21 @@           ]       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 = tryToAcquireLock 0+withAgentLock :: Agent -> IO () -> IO ()+withAgentLock agent action =+  if bootstrap agent+    then action+    else tryToAcquireLock 0   where+    agentName = name agent+     tryToAcquireLock :: Int -> IO ()     tryToAcquireLock attempts = do-      lock <- Lock.withTryLockAndPid (lockFilename name) action+      lock <- Lock.withTryLockAndPid (lockFile agent) (pidFile agent) action       when (isNothing lock) $         if attempts >= 5-          then throwIO (AgentAlreadyRunning name)+          then throwIO (AgentAlreadyRunning agentName)           else do             threadDelay (3 * 1000 * 1000)             tryToAcquireLock (attempts + 1)@@ -197,42 +225,42 @@         (verifyBootstrapSuccess agent)  verifyBootstrapSuccess :: Agent -> IO ()-verifyBootstrapSuccess Agent {name, withLog} = do+verifyBootstrapSuccess agent@(Agent {name, withLog}) = do   withLog . K.logLocM K.InfoS . K.ls $     unwords ["Waiting for another agent to take over..."] -  eAgentPid <--    Safe.tryIO $-      Retry.recoverAll-        (Retry.limitRetries 20 <> Retry.constantDelay 1000)-        (const waitForAgent)+  magentPid <- waitForAgent (Retry.limitRetries 60 <> Retry.constantDelay 1000) agent -  case eAgentPid of-    Right pid -> do+  case magentPid of+    Just pid -> do       withLog . K.logLocM K.InfoS . K.ls $         unwords ["Found an active agent for", name, "with PID " <> show pid <> ".", "Exiting."]       exitSuccess-    _ -> do+    Nothing -> do       withLog . K.logLocM K.InfoS . K.ls $         unwords ["Cannot find an active agent for", name <> ".", "Waiting for more deployments."]-  where-    lockfile = lockFilename name -    -- The PID might be stale in rare cases. Only use this for diagnostics.-    waitForAgent :: IO Posix.CPid-    waitForAgent = do-      lock <- Lock.withTryLock lockfile (pure ())-      mpid <- Lock.readPidFile lockfile-      case (lock, mpid) of-        (Nothing, Just pid) -> pure pid-        _ -> throwString "No active agent found"+waitForAgent :: Retry.RetryPolicyM IO -> Agent -> IO (Maybe Posix.CPid)+waitForAgent retryPolicy agent = do+  Retry.retrying+    retryPolicy+    (const $ pure . isNothing)+    (const $ findActiveAgent agent) -handleCommand :: Agent -> WSS.BackendCommand -> IO ()-handleCommand agent command =-  case command of-    WSS.AgentRegistered agentInformation -> registerAgent agent agentInformation-    WSS.Deployment deploymentDetails -> launchDeployment agent deploymentDetails+-- The PID might be stale in rare cases. Only use this for diagnostics.+findActiveAgent :: Agent -> IO (Maybe Posix.CPid)+findActiveAgent Agent {pidFile, lockFile} = do+  Safe.handleAny (const $ pure Nothing) $ do+    lock <- Lock.withTryLock lockFile (pure ())+    -- The lock must be held by another process+    guard (isNothing lock) +    -- We should have a PID file+    mpid <- Lock.readPidFile pidFile+    guard (isJust mpid)++    return mpid+ -- | Asynchronously open and maintain a websocket connection to the backend for -- sending deployment progress updates. connectToService :: ServiceWebSocket -> IO (IO ())@@ -271,12 +299,12 @@   | -- | Safeguard against creating root-owned files in user directories.     -- This is an issue on macOS, where, by default, sudo does not reset $HOME.     UserDoesNotOwnHome+      -- | The current user name       String-      -- ^ The current user name+      -- | The sudo user name, if any       (Maybe String)-      -- ^ The sudo user name, if any+      -- | The home directory       FilePath-      -- ^ The home directory   deriving (Show)  instance Exception AgentError where
src/Cachix/Deploy/Lock.hs view
@@ -4,6 +4,8 @@     readPidFile,     withTryLock,     withTryLockAndPid,+    lockExtension,+    pidExtension,   ) where @@ -38,22 +40,15 @@   pure lockDirectory  readPidFile :: FilePath -> IO (Maybe CPid)-readPidFile pidFilename = do-  lockDirectory <- getLockDirectory-  pidContents <- readFile (lockDirectory </> pidFilename <.> pidExtension)-  pure (readMaybe pidContents)+readPidFile pidFile = readMaybe <$> readFile pidFile --- | Run an IO action with an acquired profile lock. Returns immediately if the profile is already locked.---+-- | Run an IO action with an acquired profile lock.+-- Returns immediately if the profile is already locked. -- Lock files are not deleted after use. -- -- macOS: if using sudo, make sure to use `-H` to reset the home directory. withTryLock :: FilePath -> IO a -> IO (Maybe a)-withTryLock lockFilename action = do-  lockDirectory <- getLockDirectory--  let lockFile = lockDirectory </> lockFilename <.> lockExtension-+withTryLock lockFile action = do   bracket     (Lock.fdOpen lockFile)     (Lock.fdUnlock *> Lock.fdClose)@@ -63,13 +58,9 @@         then fmap Just action         else pure Nothing -withTryLockAndPid :: FilePath -> IO a -> IO (Maybe a)-withTryLockAndPid lockFilename action = do-  lockDirectory <- getLockDirectory--  let pidFile = lockDirectory </> lockFilename <.> pidExtension--  withTryLock lockFilename $ do+withTryLockAndPid :: FilePath -> FilePath -> IO a -> IO (Maybe a)+withTryLockAndPid lockFile pidFile action = do+  withTryLock lockFile $ do     CPid pid <- getProcessID     writeFile pidFile (show pid)     action
+ test/Daemon/PostBuildHookSpec.hs view
@@ -0,0 +1,40 @@+module Daemon.PostBuildHookSpec where++import Cachix.Client.Daemon.PostBuildHook+import Data.String+import Protolude+import qualified System.Environment as System+import Test.Hspec++spec :: Spec+spec = do+  let scriptPath = "build-hook.sh"+      configPath = "my-nix.conf"++  describe "post build hook" $ do+    it "builds the NIX_CONF environment variable" $ do+      withTempEnv ("NIX_CONF", "") $ do+        buildNixConfEnv scriptPath `shouldReturn` Nothing++      withTempEnv ("NIX_CONF", "max-jobs = 8") $ do+        buildNixConfEnv scriptPath `shouldReturn` Just ("NIX_CONF", "max-jobs = 8\npost-build-hook = " <> scriptPath)++    it "builds the NIX_USER_CONF_FILES environment variable" $ do+      withTempEnv ("NIX_USER_CONF_FILES", "") $ do+        ("NIX_USER_CONF_FILES", conf) <- buildNixUserConfFilesEnv configPath+        -- Should contain /etc/xdg and home config dirs as well+        conf `shouldContain` configPath++      withTempEnv ("NIX_USER_CONF_FILES", "/some/nix.conf") $ do+        ("NIX_USER_CONF_FILES", conf) <- buildNixUserConfFilesEnv configPath+        conf `shouldBe` intercalate ":" [configPath, "/some/nix.conf"]++withTempEnv :: (String, String) -> IO a -> IO a+withTempEnv (envName, envValue) f = bracket setEnv unsetEnv (const f)+  where+    setEnv = do+      mprevEnv <- System.lookupEnv envName+      System.setEnv envName envValue+      return mprevEnv+    unsetEnv =+      maybe (System.unsetEnv envName) (System.setEnv envName)
+ test/Daemon/PushManagerSpec.hs view
@@ -0,0 +1,122 @@+module Daemon.PushManagerSpec where++import qualified Cachix.Client.Daemon.Log as Log+import qualified Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.PushManager+import qualified Cachix.Client.Daemon.PushManager.PushJob as PushJob+import Cachix.Client.Daemon.Types.PushManager+import Cachix.Client.OptionsParser (defaultPushOptions)+import Control.Retry (defaultRetryStatus)+import qualified Data.Set as Set+import Data.Time (getCurrentTime)+import Protolude+import Test.Hspec++spec :: Spec+spec = do+  describe "push job" $ do+    it "starts in the queued state" $ do+      let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"]}+      pushJob <- PushJob.new request+      PushJob.status pushJob `shouldBe` Queued++    it "can be resolved" $ do+      let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"]}+          pathSet = Set.fromList ["foo", "bar"]+          closure = PushJob.ResolvedClosure pathSet pathSet+      initPushJob <- PushJob.new request+      timestamp <- getCurrentTime+      let pushJob = PushJob.populateQueue closure timestamp initPushJob+      PushJob.status pushJob `shouldBe` Running+      PushJob.queue pushJob `shouldBe` pathSet+      PushJob.result pushJob `shouldBe` mempty++    it "marks paths as pushed" $+      do+        let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"]}+            pathSet = Set.fromList ["foo", "bar"]+            closure = PushJob.ResolvedClosure pathSet pathSet+        timestamp <- getCurrentTime++        initPushJob <- PushJob.new request++        let pushJob =+              initPushJob+                & PushJob.populateQueue closure timestamp+                & PushJob.markStorePathPushed "foo"+        PushJob.status pushJob `shouldBe` Running+        PushJob.queue pushJob `shouldBe` Set.fromList ["bar"]+        PushJob.result pushJob+          `shouldBe` PushJob.PushResult+            { PushJob.prFailedPaths = mempty,+              PushJob.prPushedPaths = Set.fromList ["foo"],+              PushJob.prSkippedPaths = mempty+            }++    it "marks paths as failed" $+      do+        let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"]}+            pathSet = Set.fromList ["foo", "bar"]+            closure = PushJob.ResolvedClosure pathSet pathSet++        timestamp <- getCurrentTime+        initPushJob <- PushJob.new request+        let pushJob =+              initPushJob+                & PushJob.populateQueue closure timestamp+                & PushJob.markStorePathFailed "foo"+        PushJob.status pushJob `shouldBe` Running+        PushJob.queue pushJob `shouldBe` Set.fromList ["bar"]+        PushJob.result pushJob+          `shouldBe` PushJob.PushResult+            { PushJob.prFailedPaths = Set.fromList ["foo"],+              PushJob.prPushedPaths = mempty,+              PushJob.prSkippedPaths = mempty+            }++  describe "push manager" $ do+    it "queues push jobs " $ inPushManager $ do+      let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"]}+      pushId <- addPushJob request+      mpushJob <- lookupPushJob pushId+      liftIO $ do+        mpushJob `shouldSatisfy` isJust+        for_ mpushJob $ \pushJob -> do+          PushJob.pushId pushJob `shouldBe` pushId+          PushJob.pushRequest pushJob `shouldBe` request++    it "manages the lifecycle of a push job" $ inPushManager $ do+      let paths = ["bar", "foo"]++      let pushRequest = Protocol.PushRequest {Protocol.storePaths = paths}+      pushId <- addPushJob pushRequest++      let pathSet = Set.fromList paths+          closure = PushJob.ResolvedClosure pathSet pathSet+      resolvePushJob pushId closure++      withPushJob pushId $ \pushJob -> liftIO $ do+        PushJob.status pushJob `shouldBe` Running+        PushJob.startedAt pushJob `shouldSatisfy` isJust++      forM_ paths $ \path -> do+        pushStorePathAttempt path 1 defaultRetryStatus+        pushStorePathDone path++      withPushJob pushId $ \pushJob -> liftIO $ do+        PushJob.status pushJob `shouldBe` Completed+        PushJob.completedAt pushJob `shouldSatisfy` isJust+        PushJob.result pushJob+          `shouldBe` PushResult+            { prFailedPaths = mempty,+              prPushedPaths = Set.fromList paths,+              prSkippedPaths = mempty+            }++withPushManager :: (PushManagerEnv -> IO a) -> IO a+withPushManager f = do+  logger <- liftIO $ Log.new "daemon" Nothing Log.Debug+  newPushManagerEnv defaultPushOptions logger mempty >>= f++inPushManager :: PushManager a -> IO a+inPushManager f = withPushManager (`runPushManager` f)
+ test/DeploySpec.hs view
@@ -0,0 +1,60 @@+module DeploySpec where++import qualified Cachix.Client.Config as Config+import Cachix.Deploy.Agent (Agent (..), mkAgent, waitForAgent)+import Cachix.Deploy.Lock (withTryLock, withTryLockAndPid)+import qualified Cachix.Deploy.Log as Log+import qualified Cachix.Deploy.OptionsParser as CLI+import qualified Control.Retry as Retry+import Protolude+import System.IO.Temp (withSystemTempDirectory)+import Test.Hspec++spec :: Spec+spec =+  describe "deploy" $ do+    describe "lock" $ do+      it "returns Nothing if the lock is free" $+        withSystemTempDirectory "cachix-deploy-test" $ \tempDir ->+          withTestAgent tempDir $ \agent -> do+            mpid <- waitForAgent retryPolicy agent+            mpid `shouldBe` Nothing++      it "returns Nothing if there's no PID" $+        withSystemTempDirectory "cachix-deploy-test" $ \tempDir ->+          withTestAgent tempDir $ \agent -> do+            void $ withTryLock (lockFile agent) $ do+              mpid <- waitForAgent retryPolicy agent+              mpid `shouldBe` Nothing++      it "returns the PID if the lock is taken" $+        withSystemTempDirectory "cachix-deploy-test" $ \tempDir ->+          withTestAgent tempDir $ \agent -> do+            void $ withTryLockAndPid (lockFile agent) (pidFile agent) $ do+              mpid <- waitForAgent retryPolicy agent+              mpid `shouldSatisfy` isJust++withTestAgent :: FilePath -> (Agent -> IO ()) -> IO ()+withTestAgent tempDir action = do+  let logOptions =+        Log.Options+          { verbosity = Log.Verbose,+            namespace = "agent",+            environment = "Test"+          }+      agentOptions =+        CLI.AgentOptions+          { name = "foo",+            profile = Just "testing",+            bootstrap = False+          }+      agentToken = ""++  cachixOptions <- Config.defaultCachixOptions++  Log.withLog logOptions $ \withLog -> do+    agent <- mkAgent withLog logOptions (Just tempDir) cachixOptions agentOptions agentToken+    action agent++retryPolicy :: Retry.RetryPolicyM IO+retryPolicy = Retry.limitRetries 1 <> Retry.constantDelay 10