packages feed

cachix 1.7.4 → 1.7.5

raw patch · 94 files changed

+4248/−3614 lines, 94 filesdep +hnix-store-nardep ~hnix-store-core

Dependencies added: hnix-store-nar

Dependency ranges changed: hnix-store-core

Files

CHANGELOG.md view
@@ -5,11 +5,21 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [1.7.5] - 2024-10-11++### Fixed++- don't assume terminal mode if `CI=true`+- batch narinfo looksup if we have more than 1MB of hashes+- `compression-level` can't be have a short of `-c` since it clashes with `--config`+- improve hash mismatch error+- don't retry all http exceptions+ ## [1.7.4] - 2024-05-24  ### Fixed -- :ignore sigPIPE on macOS after initialising nix+- ignore sigPIPE on macOS after initialising nix  ## [1.7.3] - 2024-05-17 @@ -103,14 +113,14 @@ ### Fixed  - Reverted "Rewrite C++ bits to Haskell"-  +   This reverts 15 commits related towards getting Cachix to   built statically without C++ code in Nix.-  +   Since it's not possible to interact with Nix sqlite directly   in a reliable manner, we'll go the route of autogenerating C++   binding without Template Haskell, at some point.-  + ## [1.4.2] - 2023-04-05  ### Fixed@@ -189,7 +199,7 @@  ## [1.1] - 2022-12-16 -### Added +### Added  - Use ZSTD compresion method by default and allow overriding it via `--compression-method` back to XZ. You can also change the default permanently on your binary cache settings page. @@ -203,7 +213,7 @@  ## [1.0.1] - 2022-09-24 -### Added +### Added  - `cachix config`: allow setting hostname @@ -221,7 +231,7 @@  ## [0.8.1] - 2022-07-26 -- Cachix Deploy: retry exceptions every 1s instead of exponentially +- Cachix Deploy: retry exceptions every 1s instead of exponentially  ### Fixed @@ -241,7 +251,7 @@  ## [0.7.0] - 2022-01-12 -### Added +### Added  - Cachix Deploy support @@ -294,14 +304,14 @@  ### Fixed -- Regression: use auth token when using signing key with private caches +- Regression: use auth token when using signing key with private caches - Configure netrc even if cachix config doesn't exist  ## [0.5.0] - 2020-11-06  ### Added -- Allow specifying output directory to write nix.conf and netrc files. +- Allow specifying output directory to write nix.conf and netrc files. - Allow pushing without a Signing key using only auth token - Allow setting auth token via `$CACHIX_AUTH_TOKEN` shell variable @@ -345,7 +355,7 @@  ### Changed -- Print stderr during streaming of nars    +- Print stderr during streaming of nars  ### Fixed @@ -356,7 +366,7 @@   then shelling out (slight performance improvement)  - #251: Assert nar hash before creating narinfo-    +   Fixes rare but annoying bug of "bad nar archive" from Nix.   Never managed to reproduce. One theory is that the path disappears as   its deleted by GC or nix-store --delete.@@ -385,11 +395,11 @@  ### Fixed -- #240: push: prefer store paths as command arguments over stdin +- #240: push: prefer store paths as command arguments over stdin    Some programming languages like go and nodejs always open   stdin pipe, which confuses cachix to think there's stdin content.-  +   Sadly checking stdin for contents is tricky since we get   into the whole buffering mess.
cachix-deployment/Main.hs view
@@ -6,27 +6,27 @@ where  import Cachix.API.Error (escalateAs)-import qualified Cachix.API.WebSocketSubprotocol as AgentInformation (AgentInformation (..))-import qualified Cachix.API.WebSocketSubprotocol as DeploymentDetails (DeploymentDetails (..))-import qualified Cachix.API.WebSocketSubprotocol as WSS-import qualified Cachix.Client.URI as URI-import qualified Cachix.Deploy.Activate as Activate-import qualified Cachix.Deploy.Agent as Agent+import Cachix.API.WebSocketSubprotocol qualified as AgentInformation (AgentInformation (..))+import Cachix.API.WebSocketSubprotocol qualified as DeploymentDetails (DeploymentDetails (..))+import Cachix.API.WebSocketSubprotocol qualified as WSS+import Cachix.Client.URI qualified as URI+import Cachix.Deploy.Activate qualified as Activate+import Cachix.Deploy.Agent qualified as Agent import Cachix.Deploy.Deployment (Deployment (..))-import qualified Cachix.Deploy.Lock as Lock-import qualified Cachix.Deploy.Log as Log-import qualified Cachix.Deploy.Websocket as WebSocket-import qualified Control.Concurrent.Async as Async-import qualified Control.Concurrent.STM.TMQueue as TMQueue-import qualified Control.Exception.Safe as Safe-import qualified Data.Aeson as Aeson-import qualified Data.Conduit.TQueue as Conduit+import Cachix.Deploy.Lock qualified as Lock+import Cachix.Deploy.Log qualified as Log+import Cachix.Deploy.Websocket qualified as WebSocket+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM.TMQueue qualified as TMQueue+import Control.Exception.Safe qualified as Safe+import Data.Aeson qualified as Aeson+import Data.Conduit.TQueue qualified as Conduit import Data.Time.Clock (getCurrentTime)-import qualified Data.UUID as UUID-import qualified Data.UUID.V4 as UUID+import Data.UUID qualified as UUID+import Data.UUID.V4 qualified as UUID import GHC.IO.Encoding-import qualified Katip as K-import qualified Network.WebSockets as WS+import Katip qualified as K+import Network.WebSockets qualified as WS import Protolude hiding (toS) import Protolude.Conv import System.IO (BufferMode (..), hSetBuffering)
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.7.4+version:            1.7.5 synopsis:   Command-line client for Nix binary cache hosting https://cachix.org @@ -30,65 +30,42 @@   build-depends:      base >=4.7 && <5   default-extensions:     DeriveAnyClass-    DeriveGeneric     DerivingVia     LambdaCase-    NamedFieldPuns     NoImplicitPrelude-    NoImportQualifiedPost     OverloadedStrings     RecordWildCards-    ScopedTypeVariables    ghc-options:     -Wall -Wcompat -Wincomplete-record-updates     -Wincomplete-uni-patterns -Wredundant-constraints -fwarn-tabs     -fwarn-unused-imports -fwarn-missing-signatures-    -fwarn-name-shadowing -fwarn-incomplete-patterns+    -fwarn-name-shadowing -fwarn-incomplete-patterns -Wunused-packages+    -Wredundant-bang-patterns    -- TODO: address partial record fields in Cachix Deploy   -- if impl(ghc >= 8.4)   --   ghc-options:       -Wpartial-fields -  if impl(ghc >=8.10)-    ghc-options: -Wunused-packages--  if impl(ghc >=9.2)-    ghc-options: -Wredundant-bang-patterns+  if impl(ghc >=9.4)+    ghc-options: -Wredundant-strictness-flags -Wforall-identifier -  default-language:   Haskell2010+  default-language:   GHC2021  library   import:            defaults   exposed-modules:     Cachix.Client     Cachix.Client.CNix-    Cachix.Client.Commands-    Cachix.Client.Commands.Push+    Cachix.Client.Command+    Cachix.Client.Command.Cache+    Cachix.Client.Command.Config+    Cachix.Client.Command.Import+    Cachix.Client.Command.Pin+    Cachix.Client.Command.Push+    Cachix.Client.Command.Watch     Cachix.Client.Config     Cachix.Client.Config.Orphans-    Cachix.Client.Daemon-    Cachix.Client.Daemon.Client-    Cachix.Client.Daemon.EventLoop-    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.SocketStore-    Cachix.Client.Daemon.Subscription-    Cachix.Client.Daemon.Types-    Cachix.Client.Daemon.Types.Daemon-    Cachix.Client.Daemon.Types.EventLoop-    Cachix.Client.Daemon.Types.Log-    Cachix.Client.Daemon.Types.PushEvent-    Cachix.Client.Daemon.Types.PushManager-    Cachix.Client.Daemon.Types.SocketStore-    Cachix.Client.Daemon.Worker     Cachix.Client.Env     Cachix.Client.Exception     Cachix.Client.HumanSize@@ -106,6 +83,29 @@     Cachix.Client.URI     Cachix.Client.Version     Cachix.Client.WatchStore+    Cachix.Daemon+    Cachix.Daemon.Client+    Cachix.Daemon.EventLoop+    Cachix.Daemon.Listen+    Cachix.Daemon.Log+    Cachix.Daemon.PostBuildHook+    Cachix.Daemon.Progress+    Cachix.Daemon.Protocol+    Cachix.Daemon.Push+    Cachix.Daemon.PushManager+    Cachix.Daemon.PushManager.PushJob+    Cachix.Daemon.ShutdownLatch+    Cachix.Daemon.SocketStore+    Cachix.Daemon.Subscription+    Cachix.Daemon.Types+    Cachix.Daemon.Types.Daemon+    Cachix.Daemon.Types.Error+    Cachix.Daemon.Types.EventLoop+    Cachix.Daemon.Types.Log+    Cachix.Daemon.Types.PushEvent+    Cachix.Daemon.Types.PushManager+    Cachix.Daemon.Types.SocketStore+    Cachix.Daemon.Worker     Cachix.Deploy.Activate     Cachix.Deploy.ActivateCommand     Cachix.Deploy.Agent@@ -151,7 +151,8 @@     , generic-lens     , hercules-ci-cnix-store     , here-    , hnix-store-core         >=0.6.1.0+    , hnix-store-core         >=0.8+    , hnix-store-nar     , http-client     , http-client-tls     , http-conduit@@ -237,11 +238,12 @@ test-suite cachix-test   import:             defaults   type:               exitcode-stdio-1.0-  ghc-options:        -threaded -rtsopts -with-rtsopts=-N+  ghc-options:        -threaded -rtsopts -with-rtsopts=-N -Wno-orphans   main-is:            Main.hs   hs-source-dirs:     test   other-modules:     Daemon.PostBuildHookSpec+    Daemon.ProtocolSpec     Daemon.PushManagerSpec     DeploySpec     InstallationModeSpec@@ -261,6 +263,7 @@     , dhall     , directory     , extra+    , hercules-ci-cnix-store     , here     , hspec     , protolude
cachix/Main.hs view
@@ -1,6 +1,6 @@ module Main (main) where -import qualified Cachix.Client as CC+import Cachix.Client qualified as CC import Cachix.Client.CNix (handleCppExceptions) import Cachix.Client.Exception (CachixException) import Control.Exception.Safe (Handler (..), catches, displayException)
src/Cachix/Client.hs view
@@ -3,21 +3,27 @@   ) where -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.Command qualified as Command+import Cachix.Client.Config qualified as Config import Cachix.Client.Env (cachixoptions, mkEnv)-import Cachix.Client.OptionsParser (CachixCommand (..), DaemonCommand (..), getOpts)+import Cachix.Client.Exception (CachixException (DeprecatedCommand))+import Cachix.Client.OptionsParser+  ( CachixCommand (..),+    DaemonCommand (..),+    PushArguments (..),+    getOpts,+  ) import Cachix.Client.Version (cachixVersion)+import Cachix.Daemon qualified as Daemon+import Cachix.Daemon.Client qualified as Daemon.Client 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 Cachix.Deploy.Agent qualified as AgentCommand+import Cachix.Deploy.OptionsParser qualified as DeployOptions+import Hercules.CNix qualified as CNix+import Hercules.CNix.Util qualified as CNix.Util import Protolude import System.Console.AsciiProgress (displayConsoleRegions)-import qualified System.Posix.Signals as Signal+import System.Posix.Signals qualified as Signal  main :: IO () main = displayConsoleRegions $ do@@ -25,28 +31,30 @@   env <- mkEnv flags    initNixStore-   installSignalHandlers    let cachixOptions = cachixoptions env   case command of-    AuthToken token -> Commands.authtoken env token+    AuthToken token -> Command.authtoken env token     Config configCommand -> Config.run cachixOptions configCommand     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-    Remove name -> Commands.remove env name-    Version -> putText cachixVersion+    Daemon (DaemonWatchExec pushOptions cacheName cmd args) -> Command.watchExecDaemon env pushOptions cacheName cmd args     DeployCommand (DeployOptions.Agent opts) -> AgentCommand.run cachixOptions opts     DeployCommand (DeployOptions.Activate opts) -> ActivateCommand.run env opts+    GenerateKeypair name -> Command.generateKeypair env name+    Import pushOptions name uri -> Command.import' env pushOptions name uri+    Pin pingArgs -> Command.pin env pingArgs+    Push (PushPaths opts name cliPaths) -> Command.push env opts name cliPaths+    Push (PushWatchStore _ _) ->+      throwIO $ DeprecatedCommand "DEPRECATED: cachix watch-store has replaced cachix push --watch-store."+    Remove name -> Command.remove env name+    Use name useOptions -> Command.use env name useOptions+    Version -> putText cachixVersion+    WatchExec watchExecMode pushArgs name cmd args ->+      Command.watchExec env watchExecMode pushArgs name cmd args+    WatchStore watchArgs name -> Command.watchStore env watchArgs name  -- | Install client-wide signal handlers. installSignalHandlers :: IO ()
src/Cachix/Client/CNix.hs view
@@ -1,7 +1,7 @@ module Cachix.Client.CNix where  import Hercules.CNix.Store (Store, StorePath)-import qualified Hercules.CNix.Store as Store+import Hercules.CNix.Store qualified as Store import Language.C.Inline.Cpp.Exception import Protolude import System.Console.Pretty (Color (..), Style (..), color, style)
+ src/Cachix/Client/Command.hs view
@@ -0,0 +1,21 @@+module Cachix.Client.Command+  ( Cache.use,+    Cache.remove,+    Config.authtoken,+    Config.generateKeypair,+    Import.import',+    Pin.pin,+    Push.push,+    Watch.watchExec,+    Watch.watchExecDaemon,+    Watch.watchExecStore,+    Watch.watchStore,+  )+where++import Cachix.Client.Command.Cache qualified as Cache+import Cachix.Client.Command.Config qualified as Config+import Cachix.Client.Command.Import qualified as Import+import Cachix.Client.Command.Pin qualified as Pin+import Cachix.Client.Command.Push qualified as Push+import Cachix.Client.Command.Watch qualified as Watch
+ src/Cachix/Client/Command/Cache.hs view
@@ -0,0 +1,35 @@+module Cachix.Client.Command.Cache (use, remove) where++import Cachix.API qualified as API+import Cachix.API.Error+import Cachix.Client.Command.Push qualified as Push+import Cachix.Client.Config qualified as Config+import Cachix.Client.Env (Env (..))+import Cachix.Client.Exception (CachixException (..))+import Cachix.Client.InstallationMode qualified as InstallationMode+import Cachix.Client.NixVersion (assertNixVersion)+import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Servant+import Protolude hiding (toS)+import Servant.Auth.Client+import Servant.Client.Streaming++use :: Env -> Text -> InstallationMode.UseOptions -> IO ()+use env name useOptions = do+  optionalAuthToken <- Config.getAuthTokenMaybe (config env)+  let token = fromMaybe (Token "") optionalAuthToken+  -- 1. get cache public key+  res <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token name+  case res of+    Left err -> Push.handleCacheResponse name optionalAuthToken err+    Right binaryCache -> do+      () <- escalateAs UnsupportedNixVersion =<< assertNixVersion+      nixEnv <- InstallationMode.getNixEnv+      InstallationMode.addBinaryCache (config env) binaryCache useOptions $+        InstallationMode.getInstallationMode nixEnv useOptions++remove :: Env -> Text -> IO ()+remove env name = do+  nixEnv <- InstallationMode.getNixEnv+  InstallationMode.removeBinaryCache (Config.hostname $ config env) name $+    InstallationMode.getInstallationMode nixEnv InstallationMode.defaultUseOptions
+ src/Cachix/Client/Command/Config.hs view
@@ -0,0 +1,71 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TypeOperators #-}++module Cachix.Client.Command.Config where++import Cachix.API qualified as API+import Cachix.API.Error+import Cachix.Client.Config qualified as Config+import Cachix.Client.Env (Env (..))+import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Secrets+  ( SigningKey (SigningKey),+    exportSigningKey,+  )+import Cachix.Client.Servant+import Cachix.Types.SigningKeyCreate qualified as SigningKeyCreate+import Crypto.Sign.Ed25519 (PublicKey (PublicKey), createKeypair)+import Data.ByteString.Base64 qualified as B64+import Data.String.Here+import Data.Text qualified as T+import Data.Text.IO qualified as T.IO+import Protolude hiding (toS)+import Protolude.Conv+import Servant.API (NoContent (..))+import Servant.Auth.Client+import Servant.Client.Streaming++-- TODO: check that token actually authenticates!+authtoken :: Env -> Maybe Text -> IO ()+authtoken Env {cachixoptions} (Just token) = do+  let configPath = Config.configPath cachixoptions+  config <- Config.getConfig configPath+  Config.writeConfig configPath $ config {Config.authToken = Token (toS token)}+authtoken env Nothing = authtoken env . Just . T.strip =<< T.IO.getContents++generateKeypair :: Env -> Text -> IO ()+generateKeypair env name = do+  authToken <- Config.getAuthTokenRequired (config env)+  (PublicKey pk, sk) <- createKeypair+  let signingKey = exportSigningKey $ SigningKey sk+      signingKeyCreate = SigningKeyCreate.SigningKeyCreate (toS $ B64.encode pk)+      bcc = Config.BinaryCacheConfig name signingKey+  -- we first validate if key can be added to the binary cache+  (_ :: NoContent) <-+    escalate <=< retryHttp $+      (`runClientM` clientenv env) $+        API.createKey cachixClient authToken name signingKeyCreate+  -- if key was successfully added, write it to the config+  -- TODO: warn if binary cache with the same key already exists+  let cfg = config env & Config.setBinaryCaches [bcc]+  Config.writeConfig (Config.configPath (cachixoptions env)) cfg+  putStrLn+    ( [iTrim|+Secret signing key has been saved in the file above. To populate+your binary cache:++    $ nix-build | cachix push ${name}++Or if you'd like to use the signing key on another machine or CI:++    $ export CACHIX_SIGNING_KEY=${signingKey}+    $ nix-build | cachix push ${name}++To instruct Nix to use the binary cache:++    $ cachix use ${name}++IMPORTANT: Make sure to make a backup for the signing key above, as you have the only copy.+  |] ::+        Text+    )
+ src/Cachix/Client/Command/Import.hs view
@@ -0,0 +1,156 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE OverloadedLabels #-}++module Cachix.Client.Command.Import where++import Amazonka qualified+import Amazonka.Data.Body (ResponseBody (..))+import Amazonka.Data.Text qualified+import Amazonka.S3 qualified+import Amazonka.S3.GetObject (getObjectResponse_body)+import Amazonka.S3.ListObjectsV2 (listObjectsV2Response_contents)+import Amazonka.S3.Types.Object (object_key)+import Cachix.API.Error+import Cachix.Client.Command.Push+import Cachix.Client.Env (Env (..))+import Cachix.Client.Exception (CachixException (..))+import Cachix.Client.OptionsParser+  ( PushOptions (..),+  )+import Cachix.Client.Push+import Cachix.Client.Servant+import Cachix.Client.URI (URI)+import Cachix.Client.URI qualified as URI+import Conduit+import Control.Retry (defaultRetryStatus)+import Data.Attoparsec.Text qualified+import Data.Conduit.Combinators qualified as C+import Data.Conduit.ConcurrentMap (concurrentMapM_)+import Data.Conduit.List qualified as CL+import Data.Generics.Labels ()+import Data.Text qualified as T+import Hercules.CNix.Store (parseStorePath)+import Lens.Micro+import Network.HTTP.Client qualified as HTTP+import Network.HTTP.Types (status404)+import Nix.NarInfo qualified as NarInfo+import Protolude hiding (toS)+import Protolude.Conv+import Servant.API (NoContent (..))+import URI.ByteString qualified as UBS++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
+ src/Cachix/Client/Command/Pin.hs view
@@ -0,0 +1,44 @@+module Cachix.Client.Command.Pin (pin) where++import Cachix.API qualified as API+import Cachix.API.Error+import Cachix.Client.CNix (followLinksToStorePath)+import Cachix.Client.Config qualified as Config+import Cachix.Client.Env (Env (..))+import Cachix.Client.Exception (CachixException (..))+import Cachix.Client.OptionsParser (PinOptions (..))+import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Servant+import Cachix.Types.PinCreate qualified as PinCreate+import Data.Text qualified as T+import Hercules.CNix.Store (storePathToPath, withStore)+import Protolude hiding (toS)+import Protolude.Conv+import Servant.Client.Streaming+import System.Directory (doesFileExist)++pin :: Env -> PinOptions -> IO ()+pin env pinOpts = do+  authToken <- Config.getAuthTokenRequired (config env)+  storePath <- withStore $ \store -> do+    mpath <- followLinksToStorePath store (encodeUtf8 $ pinStorePath pinOpts)+    maybe exitFailure (storePathToPath store) mpath+  traverse_ (validateArtifact (toS storePath)) (pinArtifacts pinOpts)+  let pinCreate =+        PinCreate.PinCreate+          { name = pinName pinOpts,+            storePath = toS storePath,+            artifacts = pinArtifacts pinOpts,+            keep = pinKeep pinOpts+          }+  void $+    escalate <=< retryHttp $+      (`runClientM` clientenv env) $+        API.createPin cachixClient authToken (pinCacheName pinOpts) pinCreate+  where+    validateArtifact :: Text -> Text -> IO ()+    validateArtifact storePath artifact = do+      -- strip prefix / from artifact path if it exists+      let artifactPath = storePath <> "/" <> fromMaybe artifact (T.stripPrefix "/" artifact)+      exists <- doesFileExist (toS artifactPath)+      unless exists $ throwIO $ ArtifactNotFound $ "Artifact " <> artifactPath <> " doesn't exist."
+ src/Cachix/Client/Command/Push.hs view
@@ -0,0 +1,244 @@+{-# LANGUAGE QuasiQuotes #-}++module Cachix.Client.Command.Push+  ( push,+    pushStrategy,+    withPushParams,+    withPushParams',+    handleCacheResponse,+    getPushSecret,+    getPushSecretRequired,+  )+where++import Cachix.API qualified as API+import Cachix.Client.CNix (filterInvalidStorePath, followLinksToStorePath)+import Cachix.Client.Config qualified 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.OptionsParser as Options (PushOptions (..))+import Cachix.Client.Push as Push+import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Secrets+import Cachix.Client.Servant+import Cachix.Types.BinaryCache (BinaryCacheName)+import Cachix.Types.BinaryCache qualified as BinaryCache+import Control.Exception.Safe (throwM)+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import Control.Retry (RetryStatus (rsIterNumber))+import Data.ByteString qualified as BS+import Data.Conduit qualified as Conduit+import Data.String.Here+import Data.Text qualified 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)++push :: Env -> PushOptions -> BinaryCacheName -> [Text] -> IO ()+push env opts name cliPaths = do+  hasStdin <- not <$> hIsTerminalDevice stdin+  inputStorePaths <-+    case (hasStdin, cliPaths) of+      (False, []) -> throwIO $ NoInput "You need to specify store paths either as stdin or as an command argument"+      (True, []) -> T.words <$> getContents+      -- If we get both stdin and cli args, prefer cli args.+      -- This avoids hangs in cases where stdin is non-interactive but unused by caller+      -- some programming environments always create a (non-interactive) stdin+      -- that may or may not be written to by the caller.+      -- This is somewhat like the behavior of `cat` for example.+      (_, paths) -> return paths+  withPushParams env opts name $ \pushParams -> do+    normalized <- liftIO $+      for inputStorePaths $ \path ->+        runMaybeT $ do+          storePath <- MaybeT $ followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 path)+          MaybeT $ filterInvalidStorePath (pushParamsStore pushParams) storePath+    pushedPaths <-+      pushClosure+        (mapConcurrentlyBounded (numJobs opts))+        pushParams+        (catMaybes normalized)+    case (length normalized, length pushedPaths) of+      (0, _) -> putErrText "Nothing to push."+      (_, 0) -> putErrText "Nothing to push - all store paths are already on Cachix."+      _ -> putErrText "\nAll done."++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,+      Push.compressionMethod = compressionMethod,+      Push.compressionLevel = Options.compressionLevel opts,+      Push.chunkSize = Options.chunkSize opts,+      Push.numConcurrentChunks = Options.numConcurrentChunks opts,+      Push.omitDeriver = Options.omitDeriver opts+    }+  where+    retryText :: RetryStatus -> Text+    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 stderr+      isCI <- liftIO $ (== Just "true") <$> lookupEnv "CI"+      onTick <-+        if isTerminal && not isCI+          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 [Options.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/Command/Watch.hs view
@@ -0,0 +1,193 @@+module Cachix.Client.Command.Watch+  ( watchExec,+    watchStore,+    watchExecDaemon,+    watchExecStore,+  )+where++import Cachix.Client.Command.Push+import Cachix.Client.Env (Env (..))+import Cachix.Client.InstallationMode qualified as InstallationMode+import Cachix.Client.OptionsParser+  ( DaemonOptions (..),+    PushOptions (..),+    WatchExecMode (..),+  )+import Cachix.Client.Push+import Cachix.Client.WatchStore qualified as WatchStore+import Cachix.Daemon qualified as Daemon+import Cachix.Daemon.PostBuildHook qualified as Daemon.PostBuildHook+import Cachix.Daemon.Progress qualified as Daemon.Progress+import Cachix.Daemon.Types+import Cachix.Types.BinaryCache (BinaryCacheName)+import Conduit+import Control.Concurrent.Async qualified as Async+import Data.Conduit.Combinators qualified as C+import Data.Conduit.TMChan qualified as C+import Data.Generics.Labels ()+import Data.HashMap.Strict as HashMap+import Data.IORef+import Data.Text.IO (hGetLine)+import GHC.IO.Handle (hDuplicate, hDuplicateTo)+import Hercules.CNix.Store (withStore)+import Protolude hiding (toS)+import Protolude.Conv+import Servant.Conduit ()+import System.Console.Pretty+import System.Environment (getEnvironment)+import System.IO.Error (isEOFError)+import System.IO.Temp (withTempFile)+import System.Posix.Signals qualified as Signals+import System.Process qualified++watchStore :: Env -> PushOptions -> Text -> IO ()+watchStore env opts name = do+  withPushParams env opts name $ \pushParams ->+    WatchStore.startWorkers (pushParamsStore pushParams) (numJobs opts) pushParams++-- | Run a command and upload any new paths to the binary cache.+--+-- In auto mode, registers a post-build hook if the user is trusted.+-- Otherwise, falls back to watching the entire Nix store.+watchExec :: Env -> WatchExecMode -> PushOptions -> BinaryCacheName -> Text -> [Text] -> IO ()+watchExec env watchExecMode pushOptions cacheName cmd args = do+  case watchExecMode of+    PostBuildHook ->+      watchExecDaemon env pushOptions cacheName cmd args+    Store ->+      watchExecStore env pushOptions cacheName cmd args+    Auto -> do+      nixEnv <- InstallationMode.getNixEnv++      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."++-- | 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 $ \hookEnv ->+    withTempFile (Daemon.PostBuildHook.tempDir hookEnv) "daemon-log-capture" $ \_ logHandle ->+      withStore $ \store -> do+        let daemonOptions = DaemonOptions {daemonSocketPath = Just (Daemon.PostBuildHook.daemonSock hookEnv)}+        daemon <- Daemon.new env store daemonOptions (Just logHandle) pushOpts cacheName++        exitCode <-+          bracket (startDaemonThread daemon) (shutdownDaemonThread daemon logHandle) $ \_ -> do+            processEnv <- getEnvironment+            let newProcessEnv = Daemon.PostBuildHook.modifyEnv (Daemon.PostBuildHook.envVar hookEnv) processEnv+            let process =+                  (System.Process.proc (toS cmd) (toS <$> args))+                    { System.Process.std_out = System.Process.Inherit,+                      System.Process.env = Just newProcessEnv,+                      System.Process.delegate_ctlc = True+                    }+            System.Process.withCreateProcess process $ \_ _ _ processHandle ->+              System.Process.waitForProcess processHandle++        exitWith exitCode+  where+    -- Launch the daemon in the background and subscribe to all push events+    startDaemonThread daemon = do+      daemonThread <- Async.async $ Daemon.run daemon+      daemonChan <- Daemon.subscribe daemon+      return (daemonThread, daemonChan)++    shutdownDaemonThread daemon logHandle (daemonThread, daemonChan) = do+      -- TODO: process and fold events into a state during command execution+      daemonRes <- Async.withAsync (postWatchExec daemonChan) $ \_ -> do+        Daemon.stopIO daemon+        Async.wait daemonThread++      -- Print the daemon log in case there was an internal error+      case toExitCode daemonRes of+        ExitFailure _ -> printLog logHandle+        ExitSuccess -> return ()++    postWatchExec chan = do+      statsRef <- newIORef HashMap.empty+      runConduit $+        C.sourceTMChan chan+          .| C.mapM_ (displayPushEvent statsRef)++    -- 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 ()++    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++-- | Runs a command while watching the entire Nix store and pushing any new paths.+--+-- 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.hs 
@@ -1,494 +0,0 @@-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE OverloadedLabels #-}-{-# LANGUAGE QuasiQuotes #-}-{-# LANGUAGE TypeOperators #-}--module Cachix.Client.Commands-  ( authtoken,-    generateKeypair,-    push,-    watchStore,-    watchExec,-    watchExecDaemon,-    watchExecStore,-    use,-    import',-    remove,-    pin,-  )-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, 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 qualified Cachix.Client.InstallationMode as InstallationMode-import qualified Cachix.Client.NixConf as NixConf-import Cachix.Client.NixVersion (assertNixVersion)-import Cachix.Client.OptionsParser-  ( DaemonOptions (..),-    PinOptions (..),-    PushArguments (..),-    PushOptions (..),-  )-import Cachix.Client.Push-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 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.Monad.Trans.Maybe (MaybeT (..), runMaybeT)-import Control.Retry (defaultRetryStatus)-import Crypto.Sign.Ed25519 (PublicKey (PublicKey), createKeypair)-import qualified Data.Attoparsec.Text-import qualified Data.ByteString.Base64 as B64-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 (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.Auth.Client-import Servant.Client.Streaming-import Servant.Conduit ()-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 (withTempFile)-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 ()-authtoken Env {cachixoptions} (Just token) = do-  let configPath = Config.configPath cachixoptions-  config <- Config.getConfig configPath-  Config.writeConfig configPath $ config {Config.authToken = Token (toS token)}-authtoken env Nothing = authtoken env . Just . T.strip =<< T.IO.getContents--generateKeypair :: Env -> Text -> IO ()-generateKeypair env name = do-  authToken <- Config.getAuthTokenRequired (config env)-  (PublicKey pk, sk) <- createKeypair-  let signingKey = exportSigningKey $ SigningKey sk-      signingKeyCreate = SigningKeyCreate.SigningKeyCreate (toS $ B64.encode pk)-      bcc = Config.BinaryCacheConfig name signingKey-  -- we first validate if key can be added to the binary cache-  (_ :: NoContent) <--    escalate <=< retryHttp $-      (`runClientM` clientenv env) $-        API.createKey cachixClient authToken name signingKeyCreate-  -- if key was successfully added, write it to the config-  -- TODO: warn if binary cache with the same key already exists-  let cfg = config env & Config.setBinaryCaches [bcc]-  Config.writeConfig (Config.configPath (cachixoptions env)) cfg-  putStrLn-    ( [iTrim|-Secret signing key has been saved in the file above. To populate-your binary cache:--    $ nix-build | cachix push ${name}--Or if you'd like to use the signing key on another machine or CI:--    $ export CACHIX_SIGNING_KEY=${signingKey}-    $ nix-build | cachix push ${name}--To instruct Nix to use the binary cache:--    $ cachix use ${name}--IMPORTANT: Make sure to make a backup for the signing key above, as you have the only copy.-  |] ::-        Text-    )--getNixEnv :: IO InstallationMode.NixEnv-getNixEnv = do-  user <- InstallationMode.getUser-  nc <- NixConf.read NixConf.Global-  isTrusted <- InstallationMode.isTrustedUser $ NixConf.readLines (catMaybes [nc]) NixConf.isTrustedUsers-  isNixOS <- doesFileExist "/run/current-system/nixos-version"-  return $-    InstallationMode.NixEnv-      { InstallationMode.isRoot = user == "root",-        InstallationMode.isTrusted = isTrusted,-        InstallationMode.isNixOS = isNixOS-      }--use :: Env -> Text -> InstallationMode.UseOptions -> IO ()-use env name useOptions = do-  optionalAuthToken <- Config.getAuthTokenMaybe (config env)-  let token = fromMaybe (Token "") optionalAuthToken-  -- 1. get cache public key-  res <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token name-  case res of-    Left err -> handleCacheResponse name optionalAuthToken err-    Right binaryCache -> do-      () <- escalateAs UnsupportedNixVersion =<< assertNixVersion-      nixEnv <- getNixEnv-      InstallationMode.addBinaryCache (config env) binaryCache useOptions $-        InstallationMode.getInstallationMode nixEnv useOptions--remove :: Env -> Text -> IO ()-remove env name = do-  nixEnv <- getNixEnv-  InstallationMode.removeBinaryCache (Config.hostname $ config env) name $-    InstallationMode.getInstallationMode nixEnv InstallationMode.defaultUseOptions--push :: Env -> PushArguments -> IO ()-push env (PushPaths opts name cliPaths) = do-  hasStdin <- not <$> hIsTerminalDevice stdin-  inputStorePaths <--    case (hasStdin, cliPaths) of-      (False, []) -> throwIO $ NoInput "You need to specify store paths either as stdin or as an command argument"-      (True, []) -> T.words <$> getContents-      -- If we get both stdin and cli args, prefer cli args.-      -- This avoids hangs in cases where stdin is non-interactive but unused by caller-      -- some programming environments always create a (non-interactive) stdin-      -- that may or may not be written to by the caller.-      -- This is somewhat like the behavior of `cat` for example.-      (_, paths) -> return paths-  withPushParams env opts name $ \pushParams -> do-    normalized <- liftIO $-      for inputStorePaths $ \path ->-        runMaybeT $ do-          storePath <- MaybeT $ followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 path)-          MaybeT $ filterInvalidStorePath (pushParamsStore pushParams) storePath-    pushedPaths <--      pushClosure-        (mapConcurrentlyBounded (numJobs opts))-        pushParams-        (catMaybes normalized)-    case (length normalized, length pushedPaths) of-      (0, _) -> putErrText "Nothing to push."-      (_, 0) -> putErrText "Nothing to push - all store paths are already on Cachix."-      _ -> putErrText "\nAll done."-push _ _ =-  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-    mpath <- followLinksToStorePath store (encodeUtf8 $ pinStorePath pinOpts)-    maybe exitFailure (storePathToPath store) mpath-  traverse_ (validateArtifact (toS storePath)) (pinArtifacts pinOpts)-  let pinCreate =-        PinCreate.PinCreate-          { name = pinName pinOpts,-            storePath = toS storePath,-            artifacts = pinArtifacts pinOpts,-            keep = pinKeep pinOpts-          }-  void $-    escalate <=< retryHttp $-      (`runClientM` clientenv env) $-        API.createPin cachixClient authToken (pinCacheName pinOpts) pinCreate-  where-    validateArtifact :: Text -> Text -> IO ()-    validateArtifact storePath artifact = do-      -- strip prefix / from artifact path if it exists-      let artifactPath = storePath <> "/" <> fromMaybe artifact (T.stripPrefix "/" artifact)-      exists <- doesFileExist (toS artifactPath)-      unless exists $ throwIO $ ArtifactNotFound $ "Artifact " <> artifactPath <> " doesn't exist."--watchStore :: Env -> PushOptions -> Text -> IO ()-watchStore env opts name = do-  withPushParams env opts name $ \pushParams ->-    WatchStore.startWorkers (pushParamsStore pushParams) (numJobs opts) 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--  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."---- | 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 $ \hookEnv ->-    withTempFile (Daemon.PostBuildHook.tempDir hookEnv) "daemon-log-capture" $ \_ logHandle -> do-      let daemonOptions = DaemonOptions {daemonSocketPath = Just (Daemon.PostBuildHook.daemonSock hookEnv)}-      daemon <- Daemon.new env daemonOptions (Just logHandle) pushOpts cacheName--      exitCode <--        bracket (startDaemonThread daemon) (shutdownDaemonThread daemon logHandle) $ \_ -> do-          processEnv <- getEnvironment-          let newProcessEnv = Daemon.PostBuildHook.modifyEnv (Daemon.PostBuildHook.envVar hookEnv) processEnv-          let process =-                (System.Process.proc (toS cmd) (toS <$> args))-                  { System.Process.std_out = System.Process.Inherit,-                    System.Process.env = Just newProcessEnv,-                    System.Process.delegate_ctlc = True-                  }-          System.Process.withCreateProcess process $ \_ _ _ processHandle ->-            System.Process.waitForProcess processHandle--      exitWith exitCode-  where-    -- Launch the daemon in the background and subscribe to all push events-    startDaemonThread daemon = do-      daemonThread <- Async.async $ Daemon.run daemon-      daemonChan <- Daemon.subscribe daemon-      return (daemonThread, daemonChan)--    shutdownDaemonThread daemon logHandle (daemonThread, daemonChan) = do-      -- TODO: process and fold events into a state during command execution-      daemonExitCode <- Async.withAsync (postWatchExec daemonChan) $ \_ -> do-        Daemon.stopIO daemon-        Async.wait daemonThread--      -- Print the daemon log in case there was an internal error-      case daemonExitCode of-        ExitFailure _ -> printLog logHandle-        ExitSuccess -> return ()--    postWatchExec chan = do-      statsRef <- newIORef HashMap.empty-      runConduit $-        C.sourceTMChan chan-          .| C.mapM_ (displayPushEvent statsRef)--    -- 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 ()--    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---- | Runs a command while watching the entire Nix store and pushing any new paths.------ 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 
@@ -1,212 +0,0 @@-{-# 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 as Options-  ( PushOptions (..),-  )-import Cachix.Client.Push as 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,-      Push.compressionMethod = compressionMethod,-      Push.compressionLevel = Options.compressionLevel opts,-      Push.chunkSize = Options.chunkSize opts,-      Push.numConcurrentChunks = Options.numConcurrentChunks opts,-      Push.omitDeriver = Options.omitDeriver opts-    }-  where-    retryText :: RetryStatus -> Text-    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 [Options.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
@@ -28,14 +28,14 @@ import Cachix.Client.Config.Orphans () import Cachix.Client.Exception (CachixException (..)) import Cachix.Client.URI as URI-import qualified Control.Exception.Safe as Safe+import Control.Exception.Safe qualified as Safe import Data.Either.Extra (eitherToMaybe) import Data.String.Here-import qualified Dhall-import qualified Dhall.Pretty-import qualified Options.Applicative as Opt-import qualified Prettyprinter as Pretty-import qualified Prettyprinter.Render.Text as Pretty+import Dhall qualified+import Dhall.Pretty qualified+import Options.Applicative qualified as Opt+import Prettyprinter qualified as Pretty+import Prettyprinter.Render.Text qualified as Pretty import Protolude hiding (toS) import Protolude.Conv import Servant.Auth.Client
src/Cachix/Client/Config/Orphans.hs view
@@ -2,9 +2,9 @@  module Cachix.Client.Config.Orphans where -import qualified Data.Aeson as Aeson-import qualified Dhall-import qualified Dhall.Core+import Data.Aeson qualified as Aeson+import Dhall qualified+import Dhall.Core qualified import Protolude hiding (toS) import Protolude.Conv import Servant.Auth.Client
− src/Cachix/Client/Daemon.hs 
@@ -1,239 +0,0 @@-module Cachix.Client.Daemon-  ( Types.Daemon,-    Types.runDaemon,-    new,-    start,-    run,-    stop,-    stopIO,-    subscribe,-  )-where--import qualified Cachix.Client.Commands.Push as Commands.Push-import qualified Cachix.Client.Config as Config-import Cachix.Client.Config.Orphans ()-import qualified Cachix.Client.Daemon.EventLoop as EventLoop-import Cachix.Client.Daemon.Listen as Daemon-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 qualified Cachix.Client.Daemon.SocketStore as SocketStore-import Cachix.Client.Daemon.Subscription as Subscription-import Cachix.Client.Daemon.Types as Types-import Cachix.Client.Daemon.Types.EventLoop (DaemonEvent (ShutdownGracefully))-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 (DaemonOptions, PushOptions)-import qualified Cachix.Client.OptionsParser as Options-import Cachix.Client.Push-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.Util as CNix.Util-import qualified Katip-import qualified Network.Socket as Socket-import qualified Network.Socket.ByteString as Socket.BS-import Protolude-import System.IO.Error (isResourceVanishedError)-import System.Posix.Process (getProcessID)-import qualified System.Posix.Signals as Signal-import qualified UnliftIO.Async as Async---- | 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--  daemonSocketPath <- maybe getSocketPath pure (Options.daemonSocketPath daemonOptions)-  daemonSocketThread <- newEmptyMVar-  daemonEventLoop <- EventLoop.new-  daemonClients <- SocketStore.newSocketStore-  daemonShutdownLatch <- newShutdownLatch-  daemonPid <- getProcessID--  daemonPushSecret <- Commands.Push.getPushSecretRequired (config daemonEnv) daemonCacheName-  let authToken = getAuthTokenFromPushSecret daemonPushSecret-  daemonBinaryCache <- Push.getBinaryCache daemonEnv authToken daemonCacheName--  daemonSubscriptionManager <- Subscription.newSubscriptionManager-  let onPushEvent = Subscription.pushEvent daemonSubscriptionManager-  daemonPushManager <- PushManager.newPushManagerEnv daemonPushOptions daemonLogger onPushEvent--  return $ DaemonEnv {..}---- | Configure and run the daemon. Equivalent to running 'new' and 'run' together with some signal handling.-start :: Env -> DaemonOptions -> PushOptions -> BinaryCacheName -> IO ()-start daemonEnv daemonOptions daemonPushOptions daemonCacheName = do-  daemon <- new daemonEnv daemonOptions Nothing daemonPushOptions daemonCacheName-  installSignalHandlers daemon-  void $ run daemon---- | 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--  printConfiguration--  Push.withPushParams $ \pushParams -> do-    subscriptionManagerThread <--      Async.async $ runSubscriptionManager daemonSubscriptionManager--    let runWorkerTask =-          liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask pushParams-    workersThreads <--      Worker.startWorkers-        (Options.numJobs daemonPushOptions)-        (PushManager.pmTaskQueue daemonPushManager)-        runWorkerTask--    listenThread <- Async.async (Daemon.listen daemonEventLoop daemonSocketPath)-    liftIO $ putMVar daemonSocketThread listenThread--    EventLoop.run daemonEventLoop $ \case-      EventLoop.AddSocketClient conn ->-        SocketStore.addSocket conn (Daemon.handleClient daemonEventLoop) daemonClients-      EventLoop.RemoveSocketClient socketId ->-        SocketStore.removeSocket socketId daemonClients-      EventLoop.ReconnectSocket ->-        -- TODO: implement reconnection logic-        EventLoop.exitLoopWith (ExitFailure 1) daemonEventLoop-      EventLoop.ReceivedMessage clientMsg ->-        case clientMsg of-          ClientPushRequest pushRequest -> queueJob pushRequest-          ClientStop -> EventLoop.send daemonEventLoop EventLoop.ShutdownGracefully-          _ -> return ()-      EventLoop.ShutdownGracefully -> do-        Katip.logFM Katip.InfoS "Shutting down daemon..."-        shutdownGracefully subscriptionManagerThread workersThreads-        Katip.logFM Katip.InfoS "Daemon shut down. Exiting."-        EventLoop.exitLoopWith ExitSuccess daemonEventLoop--stop :: Daemon ()-stop = do-  eventloop <- asks daemonEventLoop-  EventLoop.send eventloop ShutdownGracefully--stopIO :: DaemonEnv -> IO ()-stopIO DaemonEnv {daemonEventLoop} =-  EventLoop.sendIO daemonEventLoop ShutdownGracefully--installSignalHandlers :: DaemonEnv -> IO ()-installSignalHandlers daemon = do-  for_ [Signal.sigTERM, Signal.sigINT] $ \signal ->-    Signal.installHandler signal (Signal.CatchOnce handler) Nothing-  where-    handler = do-      CNix.Util.triggerInterrupt-      stopIO daemon--queueJob :: Protocol.PushRequest -> Daemon ()-queueJob pushRequest = do-  daemonPushManager <- asks daemonPushManager-  -- TODO: subscribe the socket to updates if available--  -- 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 the daemon configuration to the log.-printConfiguration :: Daemon ()-printConfiguration = do-  config <- showConfiguration-  Katip.logFM Katip.InfoS $ Katip.ls $ "Configuration:\n" <> config---- | Fetch 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)-      ]--shutdownGracefully :: Async () -> [Worker.Thread] -> Daemon ()-shutdownGracefully subscriptionManagerThread workersThreads = do-  DaemonEnv {..} <- ask--  initiateShutdown daemonShutdownLatch-  PushManager.runPushManager daemonPushManager $ do-    queuedStorePathCount <- PushManager.queuedStorePathCount-    when (queuedStorePathCount > 0) $-      Katip.logFM Katip.InfoS $-        Katip.logStr $-          "Remaining store paths: " <> (show queuedStorePathCount :: Text)--  Katip.logFM Katip.DebugS "Waiting for push manager to clear remaining jobs..."-  -- Finish processing remaining push jobs-  let timeoutOptions =-        PushManager.TimeoutOptions-          { PushManager.toTimeout = 60.0,-            PushManager.toPollingInterval = 1.0-          }-  liftIO $ PushManager.stopPushManager timeoutOptions daemonPushManager-  Katip.logFM Katip.DebugS "Push manager shut down."--  -- Gracefully shut down the worker before closing the socket-  Worker.stopWorkers workersThreads--  -- Close all event subscriptions-  Katip.logFM Katip.DebugS "Shutting down event manager..."-  liftIO $ stopSubscriptionManager daemonSubscriptionManager-  Async.wait subscriptionManagerThread-  Katip.logFM Katip.DebugS "Event manager shut down."--  let sayGoodbye socket = do-        let clientSock = SocketStore.socket socket-        let clientThread = SocketStore.handlerThread socket-        Async.cancel clientThread--        -- Wave goodbye to the client that requested the shutdown-        liftIO $ Daemon.serverBye clientSock-        liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` (\_ -> return ())-        -- Wait for the other end to disconnect-        ebs <- liftIO $ try $ Socket.BS.recv clientSock 4096-        case ebs of-          Left err | isResourceVanishedError err -> Katip.logFM Katip.DebugS "Client did not disconnect cleanly."-          Left err -> Katip.logFM Katip.DebugS $ Katip.ls $ "Client socket threw an error: " <> displayException err-          Right _ -> Katip.logFM Katip.DebugS "Client disconnected."--  Async.mapConcurrently_ sayGoodbye =<< SocketStore.toList daemonClients
− src/Cachix/Client/Daemon/Client.hs 
@@ -1,132 +0,0 @@-module Cachix.Client.Daemon.Client (push, stop) where--import Cachix.Client.Daemon.Listen (getSocketPath)-import Cachix.Client.Daemon.Protocol as Protocol-import Cachix.Client.Env as Env-import Cachix.Client.OptionsParser (DaemonOptions (..))-import qualified Cachix.Client.Retry as Retry-import qualified Control.Concurrent.Async as Async-import Control.Concurrent.STM.TBMQueue-import Control.Exception.Safe (catchAny)-import qualified Data.Aeson as Aeson-import qualified Data.ByteString as BS-import qualified Data.ByteString.Lazy.Char8 as Lazy.Char8-import Data.IORef-import Data.Time.Clock-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--data SocketError-  = -- | The socket has been closed-    SocketClosed-  | -- | The socket has stopped responding to pings-    SocketStalled-  | -- | Failed to decode a message from the socket-    SocketDecodingError !Text-  deriving stock (Show)--instance Exception SocketError where-  displayException = \case-    SocketClosed -> "The socket has been closed"-    SocketStalled -> "The socket has stopped responding to pings"-    SocketDecodingError err -> "Failed to decode message from socket: " <> toS err---- | Queue up push requests with the daemon------ TODO: wait for the daemon to respond that it has received the request-push :: Env -> DaemonOptions -> [FilePath] -> IO ()-push _env daemonOptions storePaths =-  withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do-    Socket.LBS.sendAll sock $ Aeson.encode pushRequest `Lazy.Char8.snoc` '\n'-  where-    pushRequest =-      Protocol.ClientPushRequest $-        PushRequest {storePaths = storePaths}---- | Tell the daemon to stop and wait for it to gracefully exit-stop :: Env -> DaemonOptions -> IO ()-stop _env daemonOptions =-  withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do-    let size = 100-    (rx, tx) <- atomically $ (,) <$> newTBMQueue size <*> newTBMQueue size--    rxThread <- Async.async (handleIncoming rx sock)-    txThread <- Async.async (handleOutgoing tx sock)--    lastPongRef <- newIORef =<< getCurrentTime-    pingThread <- Async.async (runPingThread lastPongRef rx tx)--    mapM_ Async.link [rxThread, txThread, pingThread]--    -- Request the daemon to stop-    atomically $ writeTBMQueue tx Protocol.ClientStop--    fix $ \loop -> do-      mmsg <- atomically (readTBMQueue rx)-      case mmsg of-        Nothing -> return ()-        Just (Left err) -> putErrText $ toS $ displayException err-        Just (Right msg) ->-          case msg of-            Protocol.DaemonPong -> do-              writeIORef lastPongRef =<< getCurrentTime-              loop-            Protocol.DaemonBye -> exitSuccess-  where-    runPingThread lastPongRef rx tx = go-      where-        go = do-          timestamp <- getCurrentTime-          lastPong <- readIORef lastPongRef--          if timestamp >= addUTCTime 20 lastPong-            then atomically $ writeTBMQueue rx (Left SocketStalled)-            else do-              atomically $ writeTBMQueue tx Protocol.ClientPing-              threadDelay (2 * 1000 * 1000)-              go--    handleOutgoing tx sock = go-      where-        go = do-          mmsg <- atomically $ readTBMQueue tx-          case mmsg of-            Nothing -> return ()-            Just msg -> do-              Retry.retryAll $ const $ Socket.LBS.sendAll sock $ Aeson.encode msg `Lazy.Char8.snoc` '\n'-              go--    handleIncoming rx sock = go-      where-        go = do-          -- Wait for the socket to close-          bs <- Socket.BS.recv sock 4096 `catchAny` (\_ -> return BS.empty)--          -- A zero-length response means that the daemon has closed the socket-          if BS.null bs-            then atomically $ writeTBMQueue rx (Left SocketClosed)-            else case Aeson.eitherDecodeStrict bs of-              Left err -> do-                let terr = toS err-                putErrText terr-                atomically $ writeTBMQueue rx (Left (SocketDecodingError terr))-              Right msg -> do-                atomically $ writeTBMQueue rx (Right msg)-                go--withDaemonConn :: Maybe FilePath -> (Socket.Socket -> IO a) -> IO a-withDaemonConn optionalSocketPath f = do-  socketPath <- maybe getSocketPath pure optionalSocketPath-  bracket (open socketPath `onException` failedToConnectTo socketPath) Socket.close f-  where-    open socketPath = do-      sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol-      Retry.retryAll $ const $ Socket.connect sock (Socket.SockAddrUnix socketPath)-      return sock--    failedToConnectTo :: FilePath -> IO ()-    failedToConnectTo socketPath = do-      putErrText "\nFailed to connect to Cachix Daemon"-      putErrText $ "Tried to connect to: " <> toS socketPath <> "\n"
− src/Cachix/Client/Daemon/EventLoop.hs 
@@ -1,73 +0,0 @@-module Cachix.Client.Daemon.EventLoop-  ( new,-    send,-    sendIO,-    run,-    exitLoopWith,-    EventLoop,-    DaemonEvent (..),-  )-where--import Cachix.Client.Daemon.Types.EventLoop (DaemonEvent (..), EventLoop (..))-import Control.Concurrent.STM.TBMQueue-  ( isFullTBMQueue,-    newTBMQueueIO,-    readTBMQueue,-    tryWriteTBMQueue,-  )-import Data.Text.Lazy.Builder (toLazyText)-import qualified Katip-import Protolude--new :: (MonadIO m) => m EventLoop-new = do-  exitLatch <- liftIO newEmptyMVar-  queue <- liftIO $ newTBMQueueIO 100-  return $ EventLoop {queue, exitLatch}---- | Send an event to the event loop with logging.-send :: (Katip.KatipContext m) => EventLoop -> DaemonEvent -> m ()-send = send' Katip.logFM---- | Same as 'send', but does not require a 'Katip.KatipContext'.-sendIO :: forall m. (MonadIO m) => EventLoop -> DaemonEvent -> m ()-sendIO = send' logger-  where-    logger :: Katip.Severity -> Katip.LogStr -> m ()-    logger Katip.ErrorS msg = liftIO $ hPutStrLn stderr (toLazyText $ Katip.unLogStr msg)-    logger _ _ = return ()--send' :: (MonadIO m) => (Katip.Severity -> Katip.LogStr -> m ()) -> EventLoop -> DaemonEvent -> m ()-send' logger eventloop@(EventLoop {queue}) event = do-  res <- liftIO $ atomically $ tryWriteTBMQueue queue event-  case res of-    -- The queue is closed.-    Nothing ->-      logger Katip.DebugS "Ignored an event because the event loop is closed"-    -- Successfully wrote to the queue-    Just True -> return ()-    -- Failed to write to the queue-    Just False -> do-      isFull <- liftIO $ atomically $ isFullTBMQueue queue-      let message =-            if isFull-              then "Event loop is full"-              else "Unknown error"-      logger Katip.ErrorS $ "Failed to write to event loop: " <> message-      exitLoopWith (ExitFailure 1) eventloop--run :: (MonadIO m) => EventLoop -> (DaemonEvent -> m ()) -> m ExitCode-run eventloop f = do-  fix $ \loop -> do-    mevent <- liftIO $ atomically $ readTBMQueue (queue eventloop)-    case mevent of-      Just event -> f event-      Nothing -> exitLoopWith (ExitFailure 1) eventloop--    liftIO (tryReadMVar (exitLatch eventloop)) >>= \case-      Just exitCode -> return exitCode-      Nothing -> loop--exitLoopWith :: (MonadIO m) => ExitCode -> EventLoop -> m ()-exitLoopWith exitCode EventLoop {exitLatch} = void $ liftIO $ tryPutMVar exitLatch exitCode
− src/Cachix/Client/Daemon/Listen.hs 
@@ -1,146 +0,0 @@-module Cachix.Client.Daemon.Listen-  ( listen,-    handleClient,-    serverBye,-    getSocketPath,-    openSocket,-    closeSocket,-  )-where--import Cachix.Client.Config.Orphans ()-import qualified Cachix.Client.Daemon.EventLoop as EventLoop-import Cachix.Client.Daemon.Protocol as Protocol-import Cachix.Client.Daemon.Types.EventLoop (EventLoop)-import Cachix.Client.Daemon.Types.SocketStore (SocketId)-import Control.Exception.Safe (catchAny)-import qualified Control.Monad.Catch as E-import qualified Data.Aeson as Aeson-import qualified Data.ByteString as BS-import qualified Data.ByteString.Char8 as Char8-import qualified Katip-import Network.Socket (Socket)-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 (..),-    createDirectoryIfMissing,-    getXdgDirectory,-    removeFile,-  )-import qualified System.Environment as System-import System.FilePath ((</>))-import System.IO.Error (isDoesNotExistError, isResourceVanishedError)---- TODO: reconcile with Client-data ListenError-  = SocketError SomeException-  | DecodingError Text-  deriving stock (Show)--instance Exception ListenError where-  displayException = \case-    SocketError err -> "Failed to read from the daemon socket: " <> show err-    DecodingError err -> "Failed to decode request:\n" <> toS err---- | Listen for incoming connections on the given socket path.-listen ::-  (E.MonadMask m, Katip.KatipContext m) =>-  EventLoop ->-  FilePath ->-  m ()-listen eventloop daemonSocketPath = forever $ do-  E.bracketOnError (openSocket daemonSocketPath) closeSocket $ \sock -> do-    liftIO $ Socket.listen sock Socket.maxListenQueue-    (conn, _peerAddr) <- liftIO $ Socket.accept sock-    EventLoop.send eventloop (EventLoop.AddSocketClient conn)---- | Handle incoming messages from a client.------ Automatically responds to pings.--- Requests the daemon to remove the client socket once the loop exits.-handleClient ::-  forall m.-  (E.MonadMask m, Katip.KatipContext m) =>-  EventLoop ->-  SocketId ->-  Socket ->-  m ()-handleClient eventloop socketId conn = do-  go `E.finally` removeClient-  where-    go = do-      ebs <- liftIO $ try $ Socket.BS.recv conn 4096--      case ebs of-        Left err | isResourceVanishedError err -> return ()-        Left _ -> return ()-        -- If the socket returns 0 bytes, then it is closed-        Right bs | BS.null bs -> return ()-        Right bs -> do-          msgs <- catMaybes <$> mapM decodeMessage (Char8.split '\n' bs)--          forM_ msgs $ \msg -> do-            EventLoop.send eventloop (EventLoop.ReceivedMessage msg)-            case msg of-              Protocol.ClientPing ->-                liftIO $ Socket.LBS.sendAll conn (Aeson.encode DaemonPong)-              _ -> return ()--          go--    decodeMessage :: ByteString -> m (Maybe Protocol.ClientMessage)-    decodeMessage "" = return Nothing-    decodeMessage bs =-      case Aeson.eitherDecodeStrict bs of-        Left err -> do-          Katip.logFM Katip.ErrorS $ Katip.ls $ displayException (DecodingError (toS err))-          return Nothing-        Right msg -> return (Just msg)--    removeClient = EventLoop.send eventloop (EventLoop.RemoveSocketClient socketId)--serverBye :: Socket.Socket -> IO ()-serverBye sock =-  Socket.LBS.sendAll sock (Aeson.encode DaemonBye) `catchAny` (\_ -> return ())--getSocketPath :: IO FilePath-getSocketPath = do-  socketDir <- getSocketDir-  return $ socketDir </> "cachix-daemon.sock"--getSocketDir :: IO FilePath-getSocketDir = do-  xdgRuntimeDir <- getXdgRuntimeDir-  let socketDir = xdgRuntimeDir </> "cachix"-  createDirectoryIfMissing True socketDir-  return socketDir---- On systems with systemd: /run/user/$UID--- Otherwise, fall back to XDG_CACHE_HOME-getXdgRuntimeDir :: IO FilePath-getXdgRuntimeDir = do-  xdgRuntimeDir <- System.lookupEnv "XDG_RUNTIME_DIR"-  cacheFallback <- getXdgDirectory XdgCache ""-  return $ fromMaybe cacheFallback xdgRuntimeDir---- TODO: lock the socket-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-  -- setFileMode socketFilePath socketFileMode-  return sock-  where-    deleteSocketFileIfExists path =-      removeFile path `catch` handleDoesNotExist--    handleDoesNotExist e-      | isDoesNotExistError e = return ()-      | otherwise = throwIO e--closeSocket :: (MonadIO m) => Socket.Socket -> m ()-closeSocket = liftIO . Socket.close
− src/Cachix/Client/Daemon/Log.hs 
@@ -1,79 +0,0 @@-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, getKeys)-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))-    <> mconcat ks-    <> fromText " "-    <> Katip.unLogStr _itemMessage-  where-    nowStr = fromText (Katip.Format.formatAsLogTime _itemTime)-    ks = map brackets $ getKeys verbosity _itemPayload-    renderSeverity' severity =-      colorBySeverity withColor severity (renderSeverity severity)
− src/Cachix/Client/Daemon/PostBuildHook.hs 
@@ -1,154 +0,0 @@-{-# LANGUAGE QuasiQuotes #-}-{-# LANGUAGE TypeApplications #-}--module Cachix.Client.Daemon.PostBuildHook-  ( -- * Post-build hook setup-    PostBuildHookEnv (..),-    withSetup,--    -- * Set up env vars-    EnvVar,-    modifyEnv,--    -- * Internal-    buildNixConfEnv,-    buildNixUserConfFilesEnv,-  )-where--import Control.Monad.Catch (MonadMask)-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 (getCanonicalTemporaryDirectory, withTempDirectory)-import System.Posix.Files--type EnvVar = (String, String)--modifyEnv :: EnvVar -> [EnvVar] -> [EnvVar]-modifyEnv (envName, envValue) processEnv =-  nubOrd $ (envName, envValue) : processEnv--data PostBuildHookEnv = PostBuildHookEnv-  { tempDir :: !FilePath,-    postBuildHookConfigPath :: !FilePath,-    postBuildHookScriptPath :: !FilePath,-    daemonSock :: !FilePath,-    envVar :: !EnvVar-  }--withSetup :: Maybe FilePath -> (PostBuildHookEnv -> IO a) -> IO a-withSetup mdaemonSock f =-  withRunnerFriendlyTempDirectory "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-    envVar <- case mnixConfEnv of-      Just nixConfEnv -> return nixConfEnv-      Nothing -> do-        writeFile postBuildHookConfigPath (postBuildHookConfig postBuildHookScriptPath)-        return nixUserConfFilesEnv--    f PostBuildHookEnv {..}---- | 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-  |]---- | Run an action with a temporary directory.------ Respects the RUNNER_TEMP environment variable if set.--- This is important on self-hosted GitHub runners with locked down system temp directories.--- The directory is deleted after use.-withRunnerFriendlyTempDirectory :: (MonadIO m, MonadMask m) => String -> (FilePath -> m a) -> m a-withRunnerFriendlyTempDirectory name action = do-  runnerTempDir <- liftIO $ lookupEnv "RUNNER_TEMP"-  systemTempDir <- liftIO getCanonicalTemporaryDirectory-  let tempDir = maybe systemTempDir toS runnerTempDir-  withTempDirectory tempDir name action
− src/Cachix/Client/Daemon/Progress.hs 
@@ -1,166 +0,0 @@-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 
@@ -1,44 +0,0 @@-{-# 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-  | ClientPing-  deriving stock (Generic, Show)-  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)---- | JSON messages that the daemon can send to the client-data DaemonMessage-  = DaemonPong-  | DaemonBye-  deriving stock (Generic, Show)-  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 
@@ -1,61 +0,0 @@-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 
@@ -1,491 +0,0 @@-module Cachix.Client.Daemon.PushManager-  ( newPushManagerEnv,-    runPushManager,-    stopPushManager,--    -- * Push strategy-    newPushStrategy,--    -- * Push job-    PushJob (..),-    addPushJob,-    lookupPushJob,-    withPushJob,-    resolvePushJob,-    pendingJobCount,--    -- * Store paths-    queueStorePaths,-    removeStorePath,-    queuedStorePathCount,--    -- * Tasks-    handleTask,-    -- Push events-    pushStarted,-    pushFinished,-    pushStorePathAttempt,-    pushStorePathProgress,-    pushStorePathDone,-    pushStorePathFailed,--    -- * Helpers-    atomicallyWithTimeout,-  )-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 qualified Control.Concurrent.Async as Async-import Control.Concurrent.STM.TBMQueue-import Control.Concurrent.STM.TVar-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 (UTCTime, diffUTCTime, getCurrentTime, secondsToNominalDiffTime)-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 onPushEvent = liftIO $ do-  pmPushJobs <- newTVarIO mempty-  pmPendingJobCount <- newTVarIO 0-  pmStorePathReferences <- newTVarIO mempty-  pmTaskQueue <- newTBMQueueIO 1000-  pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)-  pmLastEventTimestamp <- newTVarIO =<< getCurrentTime-  let pmOnPushEvent id pushEvent = updateTimestampTVar pmLastEventTimestamp >> onPushEvent id pushEvent--  return $ PushManagerEnv {..}--runPushManager :: (MonadIO m) => PushManagerEnv -> PushManager a -> m a-runPushManager env f = liftIO $ unPushManager f `runReaderT` env--stopPushManager :: TimeoutOptions -> PushManagerEnv -> IO ()-stopPushManager timeoutOptions PushManagerEnv {..} = do-  atomicallyWithTimeout timeoutOptions pmLastEventTimestamp $ do-    pendingJobs <- readTVar pmPendingJobCount-    check (pendingJobs <= 0)-  atomically $ closeTBMQueue pmTaskQueue---- 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-      incrementTVar pmPendingJobCount-      writeTBMQueue pmTaskQueue $ ResolveClosure (pushId pushJob)--  return (pushId pushJob)--removePushJob :: Protocol.PushRequestId -> PushManager ()-removePushJob pushId = do-  PushManagerEnv {..} <- ask-  liftIO $ atomically $ do-    mpushJob <- HashMap.lookup pushId <$> readTVar pmPushJobs-    for_ mpushJob $ \pushJob -> do-      -- Decrement the job count if this job had not been processed yet-      unless (PushJob.isProcessed pushJob) (decrementTVar pmPendingJobCount)-      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 $ modifyPushJobSTM pushJobs pushId f--modifyPushJobSTM :: PushJobStore -> Protocol.PushRequestId -> (PushJob -> PushJob) -> STM (Maybe PushJob)-modifyPushJobSTM pushJobs pushId f =-  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--failPushJob :: Protocol.PushRequestId -> PushManager ()-failPushJob pushId = do-  PushManagerEnv {..} <- ask-  timestamp <- liftIO getCurrentTime-  liftIO $ atomically $ do-    _ <- modifyPushJobSTM pmPushJobs pushId $ PushJob.fail timestamp-    decrementTVar pmPendingJobCount--pendingJobCount :: PushManager Int-pendingJobCount = do-  pmPendingJobCount <- asks pmPendingJobCount-  liftIO $ readTVarIO pmPendingJobCount---- 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-  PushManagerEnv {..} <- ask-  mpushJob <- lookupPushJob pushId-  for_ mpushJob $ \pushJob ->-    when (Set.null $ pushQueue pushJob) $ do-      timestamp <- liftIO getCurrentTime-      liftIO $ atomically $ do-        _ <- modifyPushJobSTM pmPushJobs pushId $ PushJob.complete timestamp-        decrementTVar pmPendingJobCount-      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-  Katip.logLocM Katip.DebugS $ Katip.ls $ showClosureStats closure--  timestamp <- liftIO getCurrentTime-  _ <- modifyPushJob pushId $ PushJob.populateQueue closure timestamp--  withPushJob pushId $ \pushJob -> do-    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 ->-      runResolveClosureTask pushParams pushId-    PushStorePath filePath ->-      runPushStorePathTask pushParams filePath--runResolveClosureTask :: PushParams PushManager () -> Protocol.PushRequestId -> PushManager ()-runResolveClosureTask pushParams pushId =-  resolveClosure `withException` failJob-  where-    failJob :: SomeException -> PushManager ()-    failJob err = do-      failPushJob pushId--      Katip.katipAddContext (Katip.sl "error" (displayException err)) $-        Katip.logLocM Katip.ErrorS $-          Katip.ls $-            "Failed to resolve closure for push job " <> (show pushId :: Text)--    resolveClosure = do-      Katip.logLocM Katip.DebugS $ Katip.ls $ "Resolving closure for push job " <> (show pushId :: Text)--      withPushJob pushId $ \pushJob -> 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--runPushStorePathTask :: PushParams PushManager () -> FilePath -> PushManager ()-runPushStorePathTask pushParams filePath = do-  pushStorePath `withException` failStorePath-  where-    failStorePath =-      pushStorePathFailed filePath . toS . displayException--    pushStorePath = do-      qs <- asks pmTaskSemaphore-      E.bracket_ (QSem.waitQSem qs) (QSem.signalQSem qs) $ do-        Katip.logLocM Katip.DebugS $ Katip.ls $ "Pushing store path " <> filePath--        let store = pushParamsStore pushParams-        storePath <- liftIO $ parseStorePath store (toS filePath)--        retryAll $ uploadStorePath pushParams storePath--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.chunkSize = Client.OptionsParser.chunkSize opts,-          Client.Push.numConcurrentChunks = Client.OptionsParser.numConcurrentChunks 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--storeToFilePath :: (MonadIO m) => Store -> StorePath -> m FilePath-storeToFilePath store storePath = do-  fp <- liftIO $ storePathToPath store storePath-  pure $ toS fp---- | Canonicalize and validate a store path-normalizeStorePath :: (MonadIO m) => Store -> FilePath -> m (Maybe StorePath)-normalizeStorePath store fp =-  liftIO $ runMaybeT $ do-    storePath <- MaybeT $ followLinksToStorePath store (encodeUtf8 $ T.pack fp)-    MaybeT $ filterInvalidStorePath store storePath--withException :: (E.MonadCatch m) => m a -> (SomeException -> m a) -> m a-withException action handler = action `E.catchAll` (\e -> handler e >> E.throwM e)---- STM helpers--transactionally :: (Foldable t, MonadIO m) => t (STM ()) -> m ()-transactionally = liftIO . atomically . sequence_--updateTimestampTVar :: (MonadIO m) => TVar UTCTime -> m ()-updateTimestampTVar tvar = liftIO $ do-  now <- getCurrentTime-  atomically $ writeTVar tvar now--incrementTVar :: TVar Int -> STM ()-incrementTVar tvar = modifyTVar' tvar (+ 1)--decrementTVar :: TVar Int -> STM ()-decrementTVar tvar = modifyTVar' tvar (subtract 1)---- | Run a transaction with a timeout.-atomicallyWithTimeout ::-  TimeoutOptions ->-  -- | A TVar timestamp to compare against-  TVar UTCTime ->-  -- | The transaction to run-  STM () ->-  IO ()-atomicallyWithTimeout TimeoutOptions {..} timeVar transaction = do-  timeoutVar <- newTVarIO False-  Async.race_-    (updateShutdownTimeout timeoutVar)-    (waitForGracefulShutdown timeoutVar)-  where-    waitForGracefulShutdown timeout =-      atomically $ transaction `orElse` checkShutdownTimeout timeout--    updateShutdownTimeout timeoutVar =-      forever $ do-        now <- getCurrentTime-        atomically $ do-          timestamp <- readTVar timeVar-          let isTimeout =-                secondsToNominalDiffTime (realToFrac toTimeout) <= now `diffUTCTime` timestamp-          writeTVar timeoutVar isTimeout-        threadDelay $ ceiling (toPollingInterval * 1000.0 * 1000.0)--    checkShutdownTimeout timeout = check =<< readTVar timeout
− src/Cachix/Client/Daemon/PushManager/PushJob.hs 
@@ -1,117 +0,0 @@-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-    { pushQueue = Set.delete storePath pushQueue,-      pushResult = addPushedPath storePath pushResult-    }--markStorePathFailed :: FilePath -> PushJob -> PushJob-markStorePathFailed storePath pushJob@(PushJob {pushQueue, pushResult}) =-  pushJob-    { 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}-    }--fail :: UTCTime -> PushJob -> PushJob-fail timestamp pushJob@PushJob {..} = do-  pushJob-    { pushStatus = Failed,-      pushStats = pushStats {jsCompletedAt = Just timestamp}-    }--isCompleted :: PushJob -> Bool-isCompleted PushJob {pushStatus} = pushStatus == Completed--isFailed :: PushJob -> Bool-isFailed PushJob {pushStatus} = pushStatus == Failed--isProcessed :: PushJob -> Bool-isProcessed pushJob = isCompleted pushJob || isFailed pushJob--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 
@@ -1,26 +0,0 @@-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/SocketStore.hs 
@@ -1,43 +0,0 @@-module Cachix.Client.Daemon.SocketStore-  ( newSocketStore,-    addSocket,-    removeSocket,-    toList,-    Socket (..),-  )-where--import Cachix.Client.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..))-import Control.Concurrent.STM.TVar-import Control.Monad.IO.Unlift (MonadUnliftIO)-import qualified Data.HashMap.Strict as HashMap-import qualified Data.UUID.V4 as UUID-import qualified Network.Socket-import Protolude hiding (toList)-import qualified UnliftIO.Async as Async--newSocketStore :: (MonadIO m) => m SocketStore-newSocketStore = SocketStore <$> liftIO (newTVarIO mempty)--newSocketId :: (MonadIO m) => m SocketId-newSocketId = liftIO UUID.nextRandom--addSocket :: (MonadUnliftIO m) => Network.Socket.Socket -> (SocketId -> Network.Socket.Socket -> m ()) -> SocketStore -> m ()-addSocket socket handler (SocketStore st) = do-  socketId <- newSocketId-  handlerThread <- Async.async (handler socketId socket)-  liftIO $ atomically $ modifyTVar' st $ HashMap.insert socketId (Socket {..})--removeSocket :: (MonadIO m) => SocketId -> SocketStore -> m ()-removeSocket socketId (SocketStore stvar) = do-  msocket <- liftIO $ atomically $ stateTVar stvar $ \st ->-    let msocket = HashMap.lookup socketId st-     in (msocket, HashMap.delete socketId st)--  -- shut down the handler thread-  mapM_ (Async.uninterruptibleCancel . handlerThread) msocket--toList :: (MonadIO m) => SocketStore -> m [Socket]-toList (SocketStore st) = do-  hm <- liftIO $ readTVarIO st-  return $ HashMap.elems hm
− src/Cachix/Client/Daemon/Subscription.hs 
@@ -1,102 +0,0 @@-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 
@@ -1,26 +0,0 @@-module Cachix.Client.Daemon.Types-  ( -- * Daemon-    DaemonEnv (..),-    Daemon,-    runDaemon,--    -- * Log-    LogLevel (..),--    -- * Push-    PushManager.PushManagerEnv (..),-    PushManager.PushManager,-    PushManager.PushJob (..),-    Task (..),--    -- * Push Event-    PushEvent.PushEvent (..),-    PushEvent.PushEventMessage (..),-    PushEvent.PushRetryStatus (..),-  )-where--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 
@@ -1,93 +0,0 @@-{-# 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.EventLoop (EventLoop)-import Cachix.Client.Daemon.Types.Log (Logger)-import Cachix.Client.Daemon.Types.PushEvent (PushEvent)-import Cachix.Client.Daemon.Types.PushManager (PushManagerEnv (..))-import Cachix.Client.Daemon.Types.SocketStore (SocketStore)-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,-    -- | The main event loop-    daemonEventLoop :: EventLoop,-    -- | Push options, like compression settings and number of jobs-    daemonPushOptions :: PushOptions,-    -- | Path to the socket that the daemon listens on-    daemonSocketPath :: FilePath,-    -- | Main inbound socket thread-    daemonSocketThread :: MVar (Async ()),-    -- | 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,-    -- | Connected clients over the socket-    daemonClients :: SocketStore,-    -- | 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/EventLoop.hs 
@@ -1,35 +0,0 @@-module Cachix.Client.Daemon.Types.EventLoop-  ( EventLoop (..),-    ExitLatch,-    DaemonEvent (..),-  )-where--import qualified Cachix.Client.Daemon.Protocol as Protocol-import Cachix.Client.Daemon.Types.SocketStore (SocketId)-import Control.Concurrent.STM.TBMQueue (TBMQueue)-import Network.Socket (Socket)-import Protolude---- | An event loop that processes 'DaemonEvent's.-data EventLoop = EventLoop-  { queue :: TBMQueue DaemonEvent,-    exitLatch :: ExitLatch-  }---- | An exit latch is a semaphore that signals the event loop to exit.--- The exit code should be returned by the 'EventLoop'.-type ExitLatch = MVar ExitCode---- | Daemon events that are handled by the 'EventLoop'.-data DaemonEvent-  = -- | Shut down the daemon gracefully.-    ShutdownGracefully-  | -- | Re-establish the daemon socket-    ReconnectSocket-  | -- | Add a new client socket connection.-    AddSocketClient Socket-  | -- | Remove an existing client socket connection. For example, after it is closed.-    RemoveSocketClient SocketId-  | -- | Handle a new message from a client.-    ReceivedMessage Protocol.ClientMessage
− src/Cachix/Client/Daemon/Types/Log.hs 
@@ -1,32 +0,0 @@-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 
@@ -1,42 +0,0 @@-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-  { -- TODO: newtype a monotonic clock-    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 
@@ -1,133 +0,0 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TypeFamilies #-}--module Cachix.Client.Daemon.Types.PushManager-  ( PushManagerEnv (..),-    PushManager (..),-    PushJobStore,-    PushJob (..),-    JobStatus (..),-    JobStats (..),-    PushResult (..),-    OnPushEvent,-    Task (..),-    TimeoutOptions (..),-  )-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)---- TODO: a lot of the logic surrounding deduping, search, and job tracking could be replaced by sqlite.--- sqlite can run in-memory if we don't need persistence.--- If we do, then we can we get stop/resume for free.-data PushManagerEnv = PushManagerEnv-  { -- | A store of push jobs indexed by a PushRequestId.-    pmPushJobs :: PushJobStore,-    -- | A mapping of store paths to push requests.-    -- Used to prevent duplicate pushes and track which store paths are referenced by which push requests.-    pmStorePathReferences :: TVar (HashMap FilePath [Protocol.PushRequestId]),-    -- | FIFO queue of push tasks.-    pmTaskQueue :: TBMQueue Task,-    -- | A semaphore to control task concurrency.-    pmTaskSemaphore :: QSem,-    -- | A callback for push events.-    pmOnPushEvent :: OnPushEvent,-    -- | The timestamp of the most recent event. This is used to track activity internally.-    pmLastEventTimestamp :: TVar UTCTime,-    -- | The number of pending (uncompleted) jobs.-    pmPendingJobCount :: TVar Int,-    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)--data TimeoutOptions = TimeoutOptions-  { -- | The maximum time to wait in seconds.-    toTimeout :: Float,-    -- | The interval at which to check the timeout condition.-    toPollingInterval :: Float-  }-  deriving stock (Eq, Show)
− src/Cachix/Client/Daemon/Types/SocketStore.hs 
@@ -1,23 +0,0 @@-module Cachix.Client.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..)) where--import Control.Concurrent.STM.TVar (TVar)-import Data.HashMap.Strict (HashMap)-import Data.UUID (UUID)-import qualified Network.Socket as Network (Socket)-import Protolude--data Socket = Socket-  { socketId :: SocketId,-    socket :: Network.Socket,-    handlerThread :: Async ()-  }--instance Eq Socket where-  (==) = (==) `on` socketId--instance Ord Socket where-  compare = comparing socketId--type SocketId = UUID--newtype SocketStore = SocketStore (TVar (HashMap SocketId Socket))
− src/Cachix/Client/Daemon/Worker.hs 
@@ -1,62 +0,0 @@-module Cachix.Client.Daemon.Worker-  ( startWorkers,-    stopWorkers,-    startWorker,-    stopWorker,-    Immortal.Thread,-  )-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
@@ -3,12 +3,13 @@     mkEnv,     createClientEnv,     customManagerSettings,+    Config.defaultCachixOptions,   ) where  import Cachix.Client.Config (Config)-import qualified Cachix.Client.Config as Config-import qualified Cachix.Client.OptionsParser as Options+import Cachix.Client.Config qualified as Config+import Cachix.Client.OptionsParser qualified as Options import Cachix.Client.URI (getBaseUrl) import Cachix.Client.Version (cachixVersion) import Network.HTTP.Client
src/Cachix/Client/HumanSize.hs view
@@ -2,7 +2,7 @@  import Protolude import Text.Printf (printf)-import qualified Prelude+import Prelude qualified  humanSize :: Double -> Text humanSize size
src/Cachix/Client/InstallationMode.hs view
@@ -3,6 +3,7 @@ module Cachix.Client.InstallationMode   ( InstallationMode (..),     NixEnv (..),+    getNixEnv,     getInstallationMode,     addBinaryCache,     removeBinaryCache,@@ -16,18 +17,18 @@ where  import Cachix.Client.Config (Config)-import qualified Cachix.Client.Config as Config+import Cachix.Client.Config qualified as Config import Cachix.Client.Exception (CachixException (..))-import qualified Cachix.Client.NetRc as NetRc-import qualified Cachix.Client.NixConf as NixConf-import qualified Cachix.Client.URI as URI-import qualified Cachix.Types.BinaryCache as BinaryCache-import qualified Data.Maybe+import Cachix.Client.NetRc qualified as NetRc+import Cachix.Client.NixConf qualified as NixConf+import Cachix.Client.URI qualified as URI+import Cachix.Types.BinaryCache qualified as BinaryCache+import Data.Maybe qualified import Data.String.Here-import qualified Data.Text as T+import Data.Text qualified as T import Protolude hiding (toS) import Protolude.Conv (toS)-import System.Directory (Permissions, createDirectoryIfMissing, getPermissions, writable)+import System.Directory (Permissions, createDirectoryIfMissing, doesFileExist, getPermissions, writable) import System.Environment (lookupEnv) import System.FilePath (replaceFileName, (</>)) import System.Process (readProcessWithExitCode)@@ -76,6 +77,19 @@ toString WriteNixOS = "nixos" toString UntrustedRequiresSudo = "untrusted-requires-sudo" toString UntrustedNixOS = "untrusted-nixos"++getNixEnv :: IO NixEnv+getNixEnv = do+  user <- getUser+  nc <- NixConf.read NixConf.Global+  isTrusted <- isTrustedUser $ NixConf.readLines (catMaybes [nc]) NixConf.isTrustedUsers+  isNixOS <- doesFileExist "/run/current-system/nixos-version"+  return $+    NixEnv+      { isRoot = user == "root",+        isTrusted = isTrusted,+        isNixOS = isNixOS+      }  getInstallationMode :: NixEnv -> UseOptions -> InstallationMode getInstallationMode nixenv useOptions
src/Cachix/Client/NetRc.hs view
@@ -6,10 +6,10 @@  import Cachix.API.Error (escalateAs) import Cachix.Client.Exception (CachixException (NetRcParseError))-import qualified Cachix.Types.BinaryCache as BinaryCache-import qualified Data.ByteString as BS+import Cachix.Types.BinaryCache qualified as BinaryCache+import Data.ByteString qualified as BS import Data.List (nubBy)-import qualified Data.Text as T+import Data.Text qualified as T import Network.NetRc import Protolude hiding (toS) import Protolude.Conv
src/Cachix/Client/NixConf.hs view
@@ -33,10 +33,10 @@   ) where -import qualified Cachix.Client.URI as URI-import qualified Cachix.Types.BinaryCache as BinaryCache+import Cachix.Client.URI qualified as URI+import Cachix.Types.BinaryCache qualified as BinaryCache import Data.List (nub)-import qualified Data.Text as T+import Data.Text qualified as T import Protolude hiding (toS) import Protolude.Conv (toS) import System.Directory@@ -46,7 +46,7 @@     getXdgDirectory,   ) import System.FilePath.Posix (takeDirectory)-import qualified Text.Megaparsec as Mega+import Text.Megaparsec qualified as Mega import Text.Megaparsec.Char  data NixConfLine
src/Cachix/Client/NixVersion.hs view
@@ -12,7 +12,7 @@ import Protolude import System.Process (readProcessWithExitCode) import Text.Megaparsec (Parsec, anySingle, choice, eof, lookAhead, manyTill, parse)-import qualified Text.Megaparsec as P+import Text.Megaparsec qualified as P import Text.Megaparsec.Char (digitChar)  minimalVersion :: SemVer
src/Cachix/Client/OptionsParser.hs view
@@ -1,5 +1,3 @@-{-# LANGUAGE ApplicativeDo #-}- module Cachix.Client.OptionsParser   ( CachixCommand (..),     DaemonCommand (..),@@ -16,6 +14,9 @@     defaultNumJobs,     defaultOmitDeriver, +    -- * Watch exec+    WatchExecMode (..),+     -- * Pin options     PinOptions (..), @@ -28,20 +29,20 @@   ) where -import qualified Cachix.Client.Config as Config-import qualified Cachix.Client.InstallationMode as InstallationMode+import Cachix.Client.Config qualified as Config+import Cachix.Client.InstallationMode qualified as InstallationMode import Cachix.Client.URI (URI)-import qualified Cachix.Client.URI as URI-import qualified Cachix.Deploy.OptionsParser as DeployOptions+import Cachix.Client.URI qualified as URI+import Cachix.Deploy.OptionsParser qualified as DeployOptions import Cachix.Types.BinaryCache (BinaryCacheName)-import qualified Cachix.Types.BinaryCache as BinaryCache+import Cachix.Types.BinaryCache qualified as BinaryCache import Cachix.Types.PinCreate (Keep (..)) import Data.Conduit.ByteString (ChunkSize)-import qualified Data.Text as T+import Data.Text qualified as T import Options.Applicative import Protolude hiding (toS) import Protolude.Conv-import qualified Prelude+import Prelude qualified  data Flags = Flags   { configPath :: Config.ConfigPath,@@ -49,48 +50,6 @@     verbose :: Bool   } -flagParser :: Config.ConfigPath -> Parser Flags-flagParser defaultConfigPath = do-  hostname <--    optional $-      option-        uriOption-        ( mconcat-            [ long "hostname",-              metavar "URI",-              help ("Host to connect to (default: " <> defaultHostname <> ")")-            ]-        )-        -- Accept `host` for backwards compatibility-        <|> option uriOption (long "host" <> hidden)--  configPath <--    strOption $-      mconcat-        [ long "config",-          short 'c',-          value defaultConfigPath,-          metavar "CONFIGPATH",-          showDefault,-          help "Cachix configuration file"-        ]--  verbose <--    switch $-      mconcat-        [ long "verbose",-          short 'v',-          help "Verbose mode"-        ]--  pure Flags {hostname, configPath, verbose}-  where-    defaultHostname = URI.serialize URI.defaultCachixURI--uriOption :: ReadM URI-uriOption = eitherReader $ \s ->-  first show $ URI.parseURI (toS s)- data CachixCommand   = AuthToken (Maybe Text)   | Config Config.Command@@ -100,7 +59,7 @@   | Import PushOptions Text URI   | Pin PinOptions   | WatchStore PushOptions Text-  | WatchExec PushOptions Text Text [Text]+  | WatchExec WatchExecMode PushOptions Text Text [Text]   | Use BinaryCacheName InstallationMode.UseOptions   | Remove BinaryCacheName   | DeployCommand DeployOptions.DeployCommand@@ -125,7 +84,7 @@   { -- | The compression level to use.     compressionLevel :: Int,     -- | The compression method to use.-    -- Default value taken rom the cache settings.+    -- The default value is read from the remote binary cache settings.     compressionMethod :: Maybe BinaryCache.CompressionMethod,     -- | The size of each uploaded part.     --@@ -191,183 +150,443 @@   }   deriving (Show) +-- | CLI parser entry point+getOpts :: IO (Flags, CachixCommand)+getOpts = do+  configpath <- Config.getDefaultFilename+  let preferences = showHelpOnError <> showHelpOnEmpty <> helpShowGlobals <> subparserInline+  customExecParser (prefs preferences) (optsInfo configpath)++optsInfo :: Config.ConfigPath -> ParserInfo (Flags, CachixCommand)+optsInfo configpath =+  infoH parser $+    fullDesc+      <> progDesc "To get started, log in to https://app.cachix.org"+      <> header "https://cachix.org command line interface"+  where+    parser = (,) <$> flagParser configpath <*> (commandParser <|> versionParser)+ commandParser :: Parser CachixCommand commandParser =-  subparser $-    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 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 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 "Manage remote Nix-based systems with Cachix Deploy"))+  configCommands+    <|> cacheCommands+    <|> pushCommands+    <|> storePathCommands+    <|> daemonCommands+    <|> deployCommands   where-    nameArg = strArgument (metavar "CACHE-NAME")-    authtoken = AuthToken <$> (stdinFlag <|> (Just <$> authTokenArg))-      where-        stdinFlag = flag' Nothing (long "stdin" <> help "Read the auth token from stdin")-        authTokenArg = strArgument (metavar "AUTH-TOKEN")-    generateKeypair = GenerateKeypair <$> nameArg-    validatedLevel l =-      l <$ unless (l `elem` [0 .. 16]) (readerError $ "value " <> show l <> " not in expected range: [0..16]")-    validatedMethod :: Prelude.String -> Either Prelude.String (Maybe BinaryCache.CompressionMethod)-    validatedMethod method =+    configCommands =+      subparser $+        fold+          [ commandGroup "Config commands:",+            hidden,+            command "authtoken" $ infoH authTokenCommand $ progDesc "Configure an authentication token for Cachix",+            command "config" configCommand+          ]++    cacheCommands =+      subparser $+        fold+          [ commandGroup "Cache commands:",+            hidden,+            command "generate-keypair" $ infoH generateKeypairCommand $ progDesc "Generate a signing key pair for a binary cache",+            command "use" $ infoH useCommand $ progDesc "Configure a binary cache in nix.conf",+            command "remove" $ infoH removeCommand $ progDesc "Remove a binary cache from nix.conf"+          ]++    pushCommands =+      subparser $+        fold+          [ commandGroup "Push commands:",+            command "push" $ infoH pushCommand $ progDesc "Upload Nix store paths to a binary cache",+            command "watch-exec" $ infoH watchExecCommand $ progDesc "Run a command while watching /nix/store for newly added store paths and upload them to a binary cache",+            command "watch-store" $ infoH watchStoreCommand $ progDesc "Watch /nix/store for newly added store paths and upload them to a binary cache",+            command "import" $ infoH importCommand $ progDesc "Import the contents of a binary cache from an S3-compatible object storage service into Cachix"+          ]++    storePathCommands =+      subparser $+        fold+          [ commandGroup "Store path commands:",+            hidden,+            command "pin" $ infoH pinCommand $ progDesc "Pin a store path to prevent it from being garbage collected"+          ]++    daemonCommands =+      subparser $+        fold+          [ commandGroup "Daemon commands:",+            hidden,+            command "daemon" $ infoH daemonCommand $ progDesc "Run a daemon that listens to push requests over a unix socket"+          ]++    deployCommands =+      subparser $+        fold+          [ commandGroup "Cachix Deploy commands:",+            hidden,+            command "deploy" $ infoH deployCommand $ progDesc "Manage remote Nix-based systems with Cachix Deploy"+          ]++flagParser :: Config.ConfigPath -> Parser Flags+flagParser defaultConfigPath =+  Flags <$> configPath <*> (host <|> hostname) <*> verbose+  where+    defaultHostname = URI.serialize URI.defaultCachixURI++    hostOpts =+      [ metavar "URI",+        help $ "Host to connect to (default: " <> defaultHostname <> ")"+      ]++    -- Accept both hostname and host.+    -- TODO: switch to host.+    hostname =+      optional . option uriOption $+        long "hostname" <> mconcat hostOpts++    host =+      optional . option uriOption $+        long "host" <> metavar "URI"++    configPath =+      strOption $+        mconcat+          [ long "config",+            short 'c',+            value defaultConfigPath,+            metavar "CONFIGPATH",+            showDefault,+            help "Cachix configuration file"+          ]++    verbose =+      switch $+        long "verbose"+          <> short 'v'+          <> help "Verbose mode"++uriOption :: ReadM URI+uriOption = eitherReader (first show . URI.parseURI . toS)++cacheNameParser :: Parser BinaryCacheName+cacheNameParser = strArgument (metavar "CACHE-NAME")++authTokenCommand :: Parser CachixCommand+authTokenCommand = AuthToken <$> (stdinFlag <|> (Just <$> authTokenArg))+  where+    stdinFlag = flag' Nothing (long "stdin" <> help "Read the auth token from stdin")+    authTokenArg = strArgument (metavar "AUTH-TOKEN")++configCommand :: ParserInfo CachixCommand+configCommand = Config <$> Config.parser++generateKeypairCommand :: Parser CachixCommand+generateKeypairCommand = GenerateKeypair <$> cacheNameParser++pushOptionsParser :: Parser PushOptions+pushOptionsParser =+  PushOptions+    <$> compressionLevel+    <*> compressionMethod+    <*> chunkSize+    <*> numConcurrentChunks+    <*> numJobs+    <*> omitDeriver+  where+    compressionLevel =+      option (auto >>= validateCompressionLevel) $+        long "compression-level"+          <> short 'l'+          <> metavar "[0..16]"+          <> help+            "The compression level to use. Supported range: [0-9] for xz and [0-16] for zstd."+          <> showDefault+          <> value defaultCompressionLevel++    validateCompressionLevel l =+      l+        <$ unless+          (l `elem` [0 .. 16])+          (readerError $ "value " <> show l <> " not in expected range: [0..16]")++    compressionMethod =+      option (eitherReader validateCompressionMethod) $+        long "compression-method"+          <> short 'm'+          <> metavar "xz | zstd"+          <> help+            "The compression method to use. Overrides the preferred compression method advertised by the cache. Supported methods: xz | zstd. Defaults to zstd."+          <> value Nothing++    validateCompressionMethod :: Prelude.String -> Either Prelude.String (Maybe BinaryCache.CompressionMethod)+    validateCompressionMethod method =       if method `elem` ["xz", "zstd"]         then case readEither (T.toUpper (toS method)) of           Right a -> Right $ Just a           Left b -> Left $ toS b         else Left $ "Compression method " <> show method <> " not expected. Use xz or zstd."-    validatedChunkSize c =-      c <$ unless (c >= minChunkSize && c <= maxChunkSize) (readerError $ "value " <> show c <> " not in expected range: " <> prettyChunkRange)++    chunkSize =+      option (auto >>= validateChunkSize) $+        long "chunk-size"+          <> short 's'+          <> metavar prettyChunkRange+          <> help "The size of each uploaded part in bytes. The supported range is from 5MiB to 5GiB."+          <> showDefault+          <> value defaultChunkSize++    validateChunkSize c =+      c+        <$ unless+          (c >= minChunkSize && c <= maxChunkSize)+          (readerError $ "value " <> show c <> " not in expected range: " <> prettyChunkRange)+     prettyChunkRange = "[" <> show minChunkSize <> ".." <> show maxChunkSize <> "]"-    pushOptions :: Parser PushOptions-    pushOptions =-      PushOptions-        <$> option-          (auto >>= validatedLevel)-          ( long "compression-level"-              <> short 'c'-              <> metavar "[0..16]"-              <> help-                "The compression level to use. Supported range: [0-9] for xz and [0-16] for zstd."-              <> showDefault-              <> value defaultCompressionLevel-          )-        <*> option-          (eitherReader validatedMethod)-          ( long "compression-method"-              <> short 'm'-              <> metavar "xz | zstd"-              <> help-                "The compression method to use. Overrides the preferred compression method advertised by the cache. Supported methods: xz | zstd. Defaults to zstd."-              <> value Nothing-          )-        <*> option-          (auto >>= validatedChunkSize)-          ( long "chunk-size"-              <> short 's'-              <> metavar prettyChunkRange-              <> help "The size of each uploaded part in bytes. The supported range is from 5MiB to 5GiB."-              <> showDefault-              <> value defaultChunkSize-          )-        <*> option-          auto-          ( long "num-concurrent-chunks"-              <> short 'n'-              <> metavar "INT"-              <> help "The number of chunks to upload concurrently. The total memory usage is jobs * num-concurrent-chunks * chunk-size."-              <> showDefault-              <> value defaultNumConcurrentChunks-          )-        <*> option-          auto-          ( long "jobs"-              <> short 'j'-              <> metavar "INT"-              <> help "The number of threads to use when pushing store paths."-              <> showDefault-              <> value defaultNumJobs-          )-        <*> 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++    numConcurrentChunks =+      option auto $+        long "num-concurrent-chunks"+          <> short 'n'+          <> metavar "INT"+          <> help "The number of chunks to upload concurrently. The total memory usage is jobs * num-concurrent-chunks * chunk-size."+          <> showDefault+          <> value defaultNumConcurrentChunks++    numJobs =+      option auto $+        long "jobs"+          <> short 'j'+          <> metavar "INT"+          <> help "The number of threads to use when pushing store paths."+          <> showDefault+          <> value defaultNumJobs++    omitDeriver =+      switch $+        long "omit-deriver"+          <> help "Do not publish which derivations built the store paths."++pushCommand :: Parser CachixCommand+pushCommand = toPushCommand <$> pushOptionsParser <*> cacheNameParser <*> pushArgumentsParser+  where+    -- Figure out which subcommand we're in at the end and pass it the previous options.+    -- We can't do that beforehand because the paths argument parser needs to come after the cache name argument parser.+    toPushCommand pushOptions cacheName pushArguments =+      Push $ pushArguments pushOptions cacheName++pushArgumentsParser :: Parser (PushOptions -> BinaryCacheName -> PushArguments)+pushArgumentsParser = pushPathsParser <|> deprecatedPushWatchStoreParser++pushPathsParser :: Parser (PushOptions -> BinaryCacheName -> PushArguments)+pushPathsParser = toPushPaths <$> pathArgs+  where+    -- Move the path arguments to the end+    toPushPaths paths pushOptions cacheName =+      PushPaths pushOptions cacheName paths++    pathArgs =+      many $ strArgument (metavar "PATHS...")++deprecatedPushWatchStoreParser :: Parser (PushOptions -> BinaryCacheName -> PushArguments)+deprecatedPushWatchStoreParser =+  flag' PushWatchStore $+    long "watch-store"+      <> short 'w'+      <> help "DEPRECATED: use watch-store command instead."++importCommand :: Parser CachixCommand+importCommand =+  Import+    <$> pushOptionsParser+    <*> cacheNameParser+    <*> s3UriOption+  where+    s3UriOption =+      strArgument $+        metavar "S3-URI"+          <> help "e.g. s3://mybucket?endpoint=https://myexample.com&region=eu-central-1"++pinCommand :: Parser CachixCommand+pinCommand = Pin <$> pinOptionsParser++pinOptionsParser :: Parser PinOptions+pinOptionsParser =+  PinOptions+    <$> cacheNameParser+    <*> pinName+    <*> storePath+    <*> artifacts+    <*> keepParser+  where+    pinName = strArgument $ metavar "PIN-NAME"++    storePath = strArgument $ metavar "STORE-PATH"++    artifacts =+      many . strOption $+        metavar "ARTIFACTS..."+          <> long "artifact"+          <> short 'a'++keepParser :: Parser (Maybe Keep)+keepParser =+  daysParser+    <|> revisionsParser+    <|> foreverParser+    <|> pure Nothing+  where     -- these three flag are mutually exclusive-    daysParser = Just . Days <$> option auto (long "keep-days" <> metavar "INT")-    revisionsParser = Just . Revisions <$> option auto (long "keep-revisions" <> metavar "INT")-    foreverParser = flag' (Just Forever) (long "keep-forever")-    pinOptions =-      PinOptions-        <$> nameArg-        <*> strArgument (metavar "PIN-NAME")-        <*> strArgument (metavar "STORE-PATH")-        <*> many (strOption (metavar "ARTIFACTS..." <> long "artifact" <> short 'a'))-        <*> keepParser-    pin = Pin <$> pinOptions-    daemon =-      subparser $-        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"))-          <> 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-    pushWatchStore =-      (\() opts cache -> PushWatchStore opts cache)-        <$> flag'-          ()-          ( long "watch-store"-              <> short 'w'-              <> help "DEPRECATED: use watch-store command instead."-          )-    remove = Remove <$> nameArg-    use =-      Use-        <$> nameArg-        <*> ( InstallationMode.UseOptions-                <$> optional-                  ( option-                      (maybeReader InstallationMode.fromString)-                      ( long "mode"-                          <> short 'm'-                          <> metavar "nixos | root-nixconf | user-nixconf"-                          <> help "Mode in which to configure binary caches for Nix. Supported values: nixos | root-nixconf | user-nixconf"-                      )-                  )-                <*> strOption-                  ( long "nixos-folder"-                      <> short 'd'-                      <> help "Base directory for NixOS configuration generation"-                      <> value "/etc/nixos/"-                      <> showDefault-                  )-                <*> optional-                  ( strOption-                      ( long "output-directory"-                          <> short 'O'-                          <> help "Output directory where nix.conf and netrc will be updated."-                      )-                  )-            )+    daysParser =+      Just . Days <$> keepDays -getOpts :: IO (Flags, CachixCommand)-getOpts = do-  configpath <- Config.getDefaultFilename-  let preferences = showHelpOnError <> showHelpOnEmpty <> helpShowGlobals <> subparserInline-  customExecParser (prefs preferences) (optsInfo configpath)+    keepDays =+      option auto $+        long "keep-days"+          <> metavar "INT" -optsInfo :: Config.ConfigPath -> ParserInfo (Flags, CachixCommand)-optsInfo configpath = infoH parser desc+    revisionsParser =+      Just . Revisions <$> keepRevisions++    keepRevisions =+      option auto $+        long "keep-revisions"+          <> metavar "INT"++    foreverParser =+      flag' (Just Forever) $+        long "keep-forever"++daemonCommand :: Parser CachixCommand+daemonCommand = Daemon <$> daemonSubCommand++daemonSubCommand :: Parser DaemonCommand+daemonSubCommand =+  subparser $+    fold+      [ 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",+        command "watch-exec" $ infoH daemonWatchExec $ progDesc "Run a command and upload any store paths built during its execution"+      ]++daemonPush :: Parser DaemonCommand+daemonPush =+  DaemonPushPaths+    <$> daemonOptionsParser+    <*> many (strArgument (metavar "PATHS..."))++daemonRun :: Parser DaemonCommand+daemonRun =+  DaemonRun+    <$> daemonOptionsParser+    <*> pushOptionsParser+    <*> cacheNameParser++daemonStop :: Parser DaemonCommand+daemonStop = DaemonStop <$> daemonOptionsParser++daemonWatchExec :: Parser DaemonCommand+daemonWatchExec =+  DaemonWatchExec+    <$> pushOptionsParser+    <*> cacheNameParser+    <*> strArgument (metavar "CMD")+    <*> many (strArgument (metavar "-- ARGS"))++daemonOptionsParser :: Parser DaemonOptions+daemonOptionsParser =+  DaemonOptions <$> socketOption   where-    parser = (,) <$> flagParser configpath <*> (commandParser <|> versionParser)-    versionParser :: Parser CachixCommand-    versionParser =-      flag'-        Version-        ( long "version"-            <> short 'V'-            <> help "Show cachix version"-        )+    socketOption =+      optional . strOption $+        long "socket"+          <> short 's'+          <> metavar "SOCKET" -desc :: InfoMod a-desc =-  fullDesc-    <> progDesc "To get started log in to https://app.cachix.org"-    <> header "https://cachix.org command line interface"+deployCommand :: Parser CachixCommand+deployCommand = DeployCommand <$> DeployOptions.parser++watchExecCommand :: Parser CachixCommand+watchExecCommand =+  WatchExec+    <$> watchExecModeParser+    <*> pushOptionsParser+    <*> cacheNameParser+    <*> strArgument (metavar "CMD")+    <*> many (strArgument (metavar "-- ARGS"))++data WatchExecMode = Auto | Store | PostBuildHook deriving (Enum, Bounded)++instance Show WatchExecMode where+  show Auto = "auto"+  show Store = "store"+  show PostBuildHook = "post-build-hook"++watchExecModes :: [WatchExecMode]+watchExecModes = [minBound .. maxBound]++prettyWatchExecModes :: Prelude.String+prettyWatchExecModes = intercalate " | " (fmap show watchExecModes)++watchExecModeParser :: Parser WatchExecMode+watchExecModeParser =+  option (eitherReader parseWatchExecMode) $+    long "watch-mode"+      <> value Auto+      <> showDefault+      <> metavar prettyWatchExecModes+      <> help+        "Mode in which to watch for store paths.\+        \ \"auto\" attempts to register a post-build hook.\+        \ This pushes just the store paths built during the execution of CMD.\+        \ If that fails, it falls back to watching the entire store."++parseWatchExecMode :: Prelude.String -> Either Prelude.String WatchExecMode+parseWatchExecMode "store" = Right Store+parseWatchExecMode "post-build-hook" = Right PostBuildHook+parseWatchExecMode mode = Left $ "Unsupported mode: " <> mode <> ". Supported values: " <> prettyWatchExecModes++watchStoreCommand :: Parser CachixCommand+watchStoreCommand = WatchStore <$> pushOptionsParser <*> cacheNameParser++removeCommand :: Parser CachixCommand+removeCommand = Remove <$> cacheNameParser++useCommand :: Parser CachixCommand+useCommand = Use <$> cacheNameParser <*> installationMode++installationMode :: Parser InstallationMode.UseOptions+installationMode =+  InstallationMode.UseOptions <$> useMode <*> nixosFolderPath <*> outputDir+  where+    useMode =+      optional . option (maybeReader InstallationMode.fromString) $+        long "mode"+          <> short 'm'+          <> metavar "nixos | root-nixconf | user-nixconf"+          <> help "Mode in which to configure binary caches for Nix. Supported values: nixos | root-nixconf | user-nixconf"++    nixosFolderPath =+      strOption $+        long "nixos-folder"+          <> short 'd'+          <> help "Base directory for NixOS configuration generation"+          <> value "/etc/nixos/"+          <> showDefault++    outputDir =+      optional . strOption $+        long "output-directory"+          <> short 'O'+          <> help "Output directory where nix.conf and netrc will be updated."++versionParser :: Parser CachixCommand+versionParser =+  flag' Version $+    long "version"+      <> short 'V'+      <> help "Show cachix version"  -- TODO: usage footer infoH :: Parser a -> InfoMod a -> ParserInfo a
src/Cachix/Client/Push.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE DataKinds #-}+{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE Rank2Types #-}  {- This is a standalone module so it shouldn't depend on any CLI state like Env -}@@ -32,47 +33,51 @@     -- * Narinfo     narinfoExists,     newNarInfoCreate,+    queryNarInfoBulk,      -- * Pushing a closure of store paths     pushClosure,+    computeClosure,     getMissingPathsForClosure,     mapConcurrentlyBounded,   ) where -import qualified Cachix.API as API+import Cachix.API qualified as API import Cachix.API.Error import Cachix.API.Signing (fingerprint, passthroughHashSink, passthroughHashSinkB16, passthroughSizeSink) import Cachix.Client.Exception (CachixException (..))-import qualified Cachix.Client.Push.S3 as Push.S3+import Cachix.Client.Push.S3 qualified as Push.S3 import Cachix.Client.Retry (retryAll, retryHttp) import Cachix.Client.Secrets import Cachix.Client.Servant-import qualified Cachix.Types.BinaryCache as BinaryCache-import qualified Cachix.Types.MultipartUpload as Multipart-import qualified Cachix.Types.NarInfoCreate as Api-import qualified Cachix.Types.NarInfoHash as NarInfoHash+import Cachix.Types.BinaryCache qualified as BinaryCache+import Cachix.Types.MultipartUpload qualified as Multipart+import Cachix.Types.NarInfoCreate qualified as Api+import Cachix.Types.NarInfoHash qualified as NarInfoHash import Conduit (MonadUnliftIO) import Control.Concurrent.Async (mapConcurrently)-import qualified Control.Concurrent.QSem as QSem+import Control.Concurrent.QSem qualified as QSem import Control.Exception.Safe (MonadCatch, MonadMask, throwM) import Control.Monad.Trans.Resource (ResourceT) import Control.Retry (RetryStatus) import Crypto.Sign.Ed25519-import qualified Data.ByteString.Base64 as B64+import Data.ByteString.Base64 qualified as B64 import Data.Conduit import Data.Conduit.ByteString (ChunkSize)-import qualified Data.Conduit.Lzma as Lzma (compress)-import qualified Data.Conduit.Zstd as Zstd (compress)+import Data.Conduit.Lzma qualified as Lzma (compress)+import Data.Conduit.Zstd qualified as Zstd (compress) import Data.IORef-import qualified Data.Set as Set-import qualified Data.Text as T+import Data.List.Extra (chunksOf)+import Data.Set qualified as Set+import Data.String.Here (iTrim)+import Data.Text qualified as T import Hercules.CNix (StorePath)-import qualified Hercules.CNix.Std.Set as Std.Set+import Hercules.CNix.Std.Set qualified as Std.Set import Hercules.CNix.Store (Store)-import qualified Hercules.CNix.Store as Store+import Hercules.CNix.Store qualified as Store import Network.HTTP.Types (status401, status404)-import qualified Nix.NarInfo as NarInfo+import Nix.NarInfo qualified as NarInfo import Protolude hiding (toS) import Protolude.Conv import Servant.API@@ -80,7 +85,7 @@ import Servant.Auth.Client import Servant.Client.Streaming import Servant.Conduit ()-import qualified System.Nix.Base32+import System.Nix.Base32 qualified import System.Nix.Nar  -- | A secret for authenticating with a cache.@@ -96,7 +101,7 @@   PushSigningKey token _ -> nullTokenToMaybe token   where     nullTokenToMaybe (Token "") = Nothing-    nullTokenToMaybe (Token token) = Just $ Token token+    nullTokenToMaybe token = Just 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'.@@ -442,12 +447,22 @@   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"+      NarHashMismatch+        [iTrim|+${storePathText}: the computed NAR hash doesn't match the hash returned by the Nix store. +Expected: ${undNarHash}+Got:      ${piNarHash pathInfo}++1. Try repairing the Nix store:++   sudo nix-store --verify --repair --check-contents++2. If the issue persists, report it to https://github.com/cachix/cachix/issues+        |]+   let deriver =         fromMaybe "unknown-deriver" $           if omitDeriver strategy@@ -483,31 +498,40 @@ getMissingPathsForClosure :: (MonadIO m, MonadMask m) => PushParams n r -> [StorePath] -> m ([StorePath], [StorePath]) getMissingPathsForClosure pushParams inputPaths = do   let store = pushParamsStore pushParams-      clientEnv = pushParamsClientEnv pushParams-  -- Get the transitive closure of dependencies-  paths <- liftIO $ do-    inputs <- Std.Set.new-    for_ inputPaths $ Std.Set.insertFP inputs-    closure <- Store.computeFSClosure store Store.defaultClosureParams inputs+  paths <- computeClosure store inputPaths+  queryNarInfoBulk pushParams paths++-- | Get the transitive closure of dependencies+computeClosure :: (MonadIO m) => Store -> [StorePath] -> m [StorePath]+computeClosure store inputPaths =+  liftIO $ do+    inputSet <- Std.Set.fromListFP inputPaths+    closure <- Store.computeFSClosure store Store.defaultClosureParams inputSet     Std.Set.toListFP closure-  hashes <- liftIO $ for paths (fmap (decodeUtf8With lenientDecode) . Store.getStorePathHash)-  -- Check what store paths are missing-  missingHashesList <--    retryHttp $-      liftIO $-        escalate-          =<< API.narinfoBulk++queryNarInfoBulk :: (MonadIO m, MonadMask m) => PushParams n r -> [StorePath] -> m ([StorePath], [StorePath])+queryNarInfoBulk pushParams paths = do+  let clientEnv = pushParamsClientEnv pushParams++  pathsAndHashes <- liftIO $+    for paths $ \path -> do+      hash' <- decodeUtf8With lenientDecode <$> Store.getStorePathHash path+      pure (hash', path)+  let hashes = fmap fst pathsAndHashes+  let processBatch hashesChunk = retryHttp $ liftIO $ do+        result <-+          API.narinfoBulk             cachixClient             (getCacheAuthToken (pushParamsSecret pushParams))             (pushParamsName pushParams)-            hashes-          `runClientM` clientEnv-  let missingHashes = Set.fromList (encodeUtf8 <$> missingHashesList)-  pathsAndHashes <- liftIO $-    for paths $ \path -> do-      hash_ <- Store.getStorePathHash path-      pure (hash_, path)-  let missing = map snd $ filter (\(hash_, _path) -> Set.member hash_ missingHashes) pathsAndHashes+            hashesChunk+            `runClientM` clientEnv+        escalate result+  -- 32 bytes of each hash for 1MB limit+  missingHashesList <- concat <$> mapM processBatch (chunksOf 32768 hashes)++  let missingHashes = Set.fromList missingHashesList+  let missing = map snd $ filter (\(hash', _path) -> Set.member hash' missingHashes) pathsAndHashes   return (paths, missing)  mapConcurrentlyBounded :: (Traversable t) => Int -> (a -> IO b) -> t a -> IO (t b)
src/Cachix/Client/Push/S3.hs view
@@ -9,30 +9,30 @@   ) where -import qualified Cachix.API as API+import Cachix.API qualified as API import Cachix.API.Error import Cachix.Client.Retry (retryHttp) import Cachix.Client.Servant (cachixClient) import Cachix.Types.BinaryCache-import qualified Cachix.Types.MultipartUpload as Multipart+import Cachix.Types.MultipartUpload qualified as Multipart import Conduit (MonadResource, MonadUnliftIO) import Control.DeepSeq (rwhnf) import Crypto.Hash (Digest, MD5)-import qualified Crypto.Hash as Crypto+import Crypto.Hash qualified as Crypto import Data.ByteArray.Encoding (Base (..), convertToBase) import Data.Conduit (ConduitT, handleC, (.|)) import Data.Conduit.ByteString (ChunkSize, chunkStream)-import qualified Data.Conduit.Combinators as CC+import Data.Conduit.Combinators qualified as CC import Data.Conduit.ConcurrentMap (concurrentMapM_) import Data.List (lookup)-import qualified Data.List.NonEmpty as NonEmpty+import Data.List.NonEmpty qualified as NonEmpty import Data.UUID (UUID) import Network.HTTP.Client as HTTP-import qualified Network.HTTP.Types.Header as HTTP+import Network.HTTP.Types.Header qualified as HTTP import Protolude import Servant.Auth () import Servant.Auth.Client-import qualified Servant.Client as Client+import Servant.Client qualified as Client import Servant.Client.Streaming import Servant.Conduit ()
src/Cachix/Client/PushQueue.hs view
@@ -14,18 +14,18 @@  import Cachix.Client.CNix (filterInvalidStorePath) import Cachix.Client.Push (getMissingPathsForClosure)-import qualified Cachix.Client.Push as Push+import Cachix.Client.Push qualified as Push import Cachix.Client.Retry (retryAll) import Control.Concurrent.Async import Control.Concurrent.Extra (once) import Control.Concurrent.STM (TVar, modifyTVar', newTVarIO, readTVar)-import qualified Control.Concurrent.STM.Lock as Lock-import qualified Control.Concurrent.STM.TBQueue as TBQueue-import qualified Data.Set as S+import Control.Concurrent.STM.Lock qualified as Lock+import Control.Concurrent.STM.TBQueue qualified as TBQueue+import Data.Set qualified as S import Hercules.CNix.Store (StorePath) import Protolude-import qualified System.Posix.Signals as Signals-import qualified System.Systemd.Daemon as Systemd+import System.Posix.Signals qualified as Signals+import System.Systemd.Daemon qualified as Systemd  type Queue = TBQueue.TBQueue StorePath
src/Cachix/Client/Retry.hs view
@@ -1,7 +1,6 @@ module Cachix.Client.Retry   ( retryAll,     retryAllWithPolicy,-    retryAllWithLogging,     retryHttp,     retryHttpWith,     endlessRetryPolicy,@@ -10,11 +9,8 @@ where  import Cachix.Client.Exception (CachixException (..))-import qualified Control.Concurrent.Async as Async-import Control.Exception.Safe-  ( Handler (..),-    isSyncException,-  )+import Control.Concurrent.Async qualified as Async+import Control.Exception.Safe (Handler (..)) import Control.Monad.Catch (MonadCatch, MonadMask, handleJust, throwM) import Control.Retry import Data.List (lookup)@@ -28,11 +24,13 @@     rfc822DateFormat,   ) import GHC.Read (Read (readPrec))-import qualified Network.HTTP.Client as HTTP+import Network.HTTP.Client qualified as HTTP import Network.HTTP.Types.Header (hRetryAfter)-import qualified Network.HTTP.Types.Status as HTTP+import Network.HTTP.Types.Status qualified as HTTP import Protolude hiding (Handler (..), handleJust)-import qualified Text.ParserCombinators.ReadPrec as ReadPrec (lift)+import Servant.Client (ClientError (..))+import Servant.Client qualified as Servant+import Text.ParserCombinators.ReadPrec qualified as ReadPrec (lift)  defaultRetryPolicy :: RetryPolicy defaultRetryPolicy =@@ -65,25 +63,6 @@     -- 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 =@@ -105,20 +84,27 @@   where     handlers :: [RetryStatus -> Handler m RetryAction]     handlers =-      skipAsyncExceptions' ++ [retryHttpExceptions, retrySyncExceptions]+      skipAsyncExceptions' ++ [retryHttpExceptions, retryClientExceptions]      skipAsyncExceptions' = map (fmap toRetryAction .) skipAsyncExceptions -    retryHttpExceptions _ = Handler httpExceptionToRetryAction-    retrySyncExceptions _ = Handler $ \(_ :: SomeException) -> return ConsultPolicy+    retryHttpExceptions _ = Handler httpExceptionToRetryHandler+    retryClientExceptions _ = Handler clientExceptionToRetryHandler -    httpExceptionToRetryAction :: HTTP.HttpException -> m RetryAction-    httpExceptionToRetryAction (HTTP.HttpExceptionRequest _ (HTTP.StatusCodeException response _))+    httpExceptionToRetryHandler :: HTTP.HttpException -> m RetryAction+    httpExceptionToRetryHandler (HTTP.HttpExceptionRequest _ (HTTP.StatusCodeException response _))       | statusMayHaveRetryHeader (HTTP.responseStatus response) = overrideDelayWithRetryAfter response-    httpExceptionToRetryAction ex = return . toRetryAction . shouldRetryHttpException $ ex+    httpExceptionToRetryHandler ex = return . toRetryAction . shouldRetryHttpException $ ex -    statusMayHaveRetryHeader :: HTTP.Status -> Bool-    statusMayHaveRetryHeader = flip elem [HTTP.tooManyRequests429, HTTP.serviceUnavailable503]+    clientExceptionToRetryHandler :: ClientError -> m RetryAction+    clientExceptionToRetryHandler (FailureResponse _req res)+      | shouldRetryHttpStatusCode (Servant.responseStatusCode res) =+          return ConsultPolicy+    clientExceptionToRetryHandler (ConnectionError ex) =+      case fromException ex of+        Just httpException -> httpExceptionToRetryHandler httpException+        Nothing -> return DontRetry+    clientExceptionToRetryHandler _ = return DontRetry  data RetryAfter   = RetryAfterDate UTCTime@@ -171,9 +157,16 @@       | 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.StatusCodeException response _ ->+      shouldRetryHttpStatusCode (HTTP.responseStatus response)     HTTP.HttpZlibException _ -> True     _ -> False++-- | Determine whether the HTTP status code is worth retrying.+shouldRetryHttpStatusCode :: HTTP.Status -> Bool+shouldRetryHttpStatusCode code | code == HTTP.tooManyRequests429 = True+shouldRetryHttpStatusCode code | HTTP.statusIsServerError code = True+shouldRetryHttpStatusCode _ = False++statusMayHaveRetryHeader :: HTTP.Status -> Bool+statusMayHaveRetryHeader = flip elem [HTTP.tooManyRequests429, HTTP.serviceUnavailable503]
src/Cachix/Client/Secrets.hs view
@@ -10,8 +10,8 @@  -- TODO: * Auth token import Crypto.Sign.Ed25519-import qualified Data.ByteString.Base64 as B64-import qualified Data.ByteString.Char8 as BC+import Data.ByteString.Base64 qualified as B64+import Data.ByteString.Char8 qualified as BC import Protolude hiding (toS) import Protolude.Conv
src/Cachix/Client/Servant.hs view
@@ -10,15 +10,15 @@   ) where -import qualified Cachix.API as API-import qualified Cachix.API.Deploy.V1 as API.Deploy.V1-import qualified Cachix.API.Deploy.V2 as API.Deploy.V2+import Cachix.API qualified as API+import Cachix.API.Deploy.V1 qualified as API.Deploy.V1+import Cachix.API.Deploy.V2 qualified as API.Deploy.V2 import Cachix.Types.ContentTypes () import Network.HTTP.Types (Status) import Protolude import Servant.API.Generic import Servant.Auth.Client ()-import qualified Servant.Client+import Servant.Client qualified import Servant.Client.Streaming import Servant.Conduit ()
src/Cachix/Client/URI.hs view
@@ -25,20 +25,20 @@ where  import Control.Monad (fail)-import qualified Data.Aeson as Aeson-import qualified Data.ByteString as BS-import qualified Data.ByteString.Char8 as Char8+import Data.Aeson qualified as Aeson+import Data.ByteString qualified as BS+import Data.ByteString.Char8 qualified as Char8 import Data.Either.Validation (Validation (Failure, Success))-import qualified Data.Map.Strict as Map+import Data.Map.Strict qualified as Map import Data.Maybe (fromJust) import Data.String-import qualified Dhall-import qualified Dhall.Core+import Dhall qualified+import Dhall.Core qualified import Protolude hiding (toS) import Protolude.Conv import Servant.Client-import qualified URI.ByteString as UBS-import qualified URI.ByteString.QQ as UBS+import URI.ByteString qualified as UBS+import URI.ByteString.QQ qualified as UBS  -- Default URIs
src/Cachix/Client/WatchStore.hs view
@@ -5,13 +5,13 @@  import Cachix.Client.CNix (filterInvalidStorePath) import Cachix.Client.Push-import qualified Cachix.Client.PushQueue as PushQueue-import qualified Control.Concurrent.STM.TBQueue as TBQueue+import Cachix.Client.PushQueue qualified as PushQueue+import Control.Concurrent.STM.TBQueue qualified as TBQueue import Hercules.CNix.Store (Store)-import qualified Hercules.CNix.Store as Store+import Hercules.CNix.Store qualified as Store import Protolude import System.FSNotify-import qualified System.Systemd.Daemon as Systemd+import System.Systemd.Daemon qualified as Systemd  startWorkers :: Store -> Int -> PushParams IO () -> IO () startWorkers store numWorkers pushParams = do
+ src/Cachix/Daemon.hs view
@@ -0,0 +1,283 @@+module Cachix.Daemon+  ( Types.Daemon,+    Types.runDaemon,+    new,+    start,+    run,+    stop,+    stopIO,+    subscribe,+  )+where++import Cachix.Client.Command.Push qualified as Command.Push+import Cachix.Client.Config qualified as Config+import Cachix.Client.Config.Orphans ()+import Cachix.Client.Env as Env+import Cachix.Client.OptionsParser (DaemonOptions, PushOptions)+import Cachix.Client.OptionsParser qualified as Options+import Cachix.Client.Push+import Cachix.Daemon.EventLoop qualified as EventLoop+import Cachix.Daemon.Listen as Listen+import Cachix.Daemon.Log qualified as Log+import Cachix.Daemon.Protocol as Protocol+import Cachix.Daemon.Push as Push+import Cachix.Daemon.PushManager qualified as PushManager+import Cachix.Daemon.ShutdownLatch+import Cachix.Daemon.SocketStore qualified as SocketStore+import Cachix.Daemon.Subscription as Subscription+import Cachix.Daemon.Types as Types+import Cachix.Daemon.Types.PushManager qualified as PushManager+import Cachix.Daemon.Worker qualified as Worker+import Cachix.Types.BinaryCache (BinaryCacheName)+import Cachix.Types.BinaryCache qualified as BinaryCache+import Control.Concurrent.STM.TMChan+import Control.Exception.Safe (catchAny)+import Data.Text qualified as T+import Hercules.CNix.Store (Store, withStore)+import Hercules.CNix.Util qualified as CNix.Util+import Katip qualified+import Network.Socket qualified as Socket+import Network.Socket.ByteString qualified as Socket.BS+import Protolude hiding (bracket)+import System.IO.Error (isResourceVanishedError)+import System.Posix.Process (getProcessID)+import System.Posix.Signals qualified as Signal+import UnliftIO (MonadUnliftIO)+import UnliftIO.Async qualified as Async+import UnliftIO.Exception (bracket)++-- | Configure a new daemon. Use 'run' to start it.+new ::+  -- | The Cachix environment.+  Env ->+  -- | A handle to the Nix store.+  Store ->+  -- | 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 nixStore daemonOptions daemonLogHandle daemonPushOptions daemonCacheName = do+  let daemonLogLevel =+        if Config.verbose (Env.cachixoptions daemonEnv)+          then Debug+          else Info+  daemonLogger <- Log.new "cachix.daemon" daemonLogHandle daemonLogLevel++  daemonEventLoop <- EventLoop.new+  daemonShutdownLatch <- newShutdownLatch+  daemonPid <- getProcessID++  daemonSocketPath <- maybe getSocketPath pure (Options.daemonSocketPath daemonOptions)+  daemonSocketThread <- newEmptyMVar+  daemonClients <- SocketStore.newSocketStore++  daemonPushSecret <- Command.Push.getPushSecretRequired (config daemonEnv) daemonCacheName+  let authToken = getAuthTokenFromPushSecret daemonPushSecret+  daemonBinaryCache <- Push.getBinaryCache daemonEnv authToken daemonCacheName++  daemonSubscriptionManagerThread <- newEmptyMVar+  daemonSubscriptionManager <- Subscription.newSubscriptionManager+  let onPushEvent = Subscription.pushEvent daemonSubscriptionManager++  let pushParams = Push.newPushParams nixStore (clientenv daemonEnv) daemonBinaryCache daemonPushSecret daemonPushOptions+  daemonPushManager <- PushManager.newPushManagerEnv daemonPushOptions pushParams onPushEvent daemonLogger++  daemonWorkerThreads <- newEmptyMVar++  return $ DaemonEnv {..}++-- | Configure and run the daemon as a CLI command.+-- Equivalent to running 'withStore', new', and 'run', together with some signal handling.+start :: Env -> DaemonOptions -> PushOptions -> BinaryCacheName -> IO ()+start daemonEnv daemonOptions daemonPushOptions daemonCacheName =+  withStore $ \store -> do+    daemon <- new daemonEnv store daemonOptions Nothing daemonPushOptions daemonCacheName+    installSignalHandlers daemon+    result <- run daemon+    exitWith (toExitCode result)++-- | Run a daemon from a given configuration.+run :: DaemonEnv -> IO (Either DaemonError ())+run daemon = fmap join <$> runDaemon daemon $ do+  Katip.logFM Katip.InfoS "Starting Cachix Daemon"+  DaemonEnv {..} <- ask++  printConfiguration++  Async.async (runSubscriptionManager daemonSubscriptionManager)+    >>= (liftIO . putMVar daemonSubscriptionManagerThread)++  Worker.startWorkers+    (Options.numJobs daemonPushOptions)+    (PushManager.pmTaskQueue daemonPushManager)+    (liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask)+    >>= (liftIO . putMVar daemonWorkerThreads)++  Async.async (Listen.listen daemonEventLoop daemonSocketPath)+    >>= (liftIO . putMVar daemonSocketThread)++  eventLoopRes <- EventLoop.run daemonEventLoop $ \case+    AddSocketClient conn ->+      SocketStore.addSocket conn (Listen.handleClient daemonEventLoop) daemonClients+    RemoveSocketClient socketId ->+      SocketStore.removeSocket socketId daemonClients+    ReconnectSocket ->+      -- TODO: implement reconnection logic+      EventLoop.exitLoopWith (Left DaemonSocketError) daemonEventLoop+    ReceivedMessage clientMsg ->+      case clientMsg of+        ClientPushRequest pushRequest -> queueJob pushRequest+        ClientStop -> EventLoop.send daemonEventLoop ShutdownGracefully+        _ -> return ()+    ShutdownGracefully -> do+      Katip.logFM Katip.InfoS "Shutting down daemon..."+      pushResult <- shutdownGracefully+      Katip.logFM Katip.InfoS "Daemon shut down. Exiting."+      EventLoop.exitLoopWith pushResult daemonEventLoop++  return $ case eventLoopRes of+    Left err -> Left (DaemonEventLoopError err)+    Right (Left err) -> Left err+    Right (Right ()) -> Right ()++stop :: Daemon ()+stop = do+  eventloop <- asks daemonEventLoop+  EventLoop.send eventloop ShutdownGracefully++stopIO :: DaemonEnv -> IO ()+stopIO DaemonEnv {daemonEventLoop} =+  EventLoop.sendIO daemonEventLoop ShutdownGracefully++installSignalHandlers :: DaemonEnv -> IO ()+installSignalHandlers daemon = do+  for_ [Signal.sigTERM, Signal.sigINT] $ \signal ->+    Signal.installHandler signal (Signal.CatchOnce handler) Nothing+  where+    handler = do+      CNix.Util.triggerInterrupt+      stopIO daemon++queueJob :: Protocol.PushRequest -> Daemon ()+queueJob pushRequest = do+  daemonPushManager <- asks daemonPushManager+  -- TODO: subscribe the socket to updates if available++  -- 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 the daemon configuration to the log.+printConfiguration :: Daemon ()+printConfiguration = do+  config <- showConfiguration+  Katip.logFM Katip.InfoS $ Katip.ls $ "Configuration:\n" <> config++-- | Fetch 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)+      ]++shutdownGracefully :: Daemon (Either DaemonError ())+shutdownGracefully = do+  DaemonEnv {..} <- ask++  -- Indicate that the daemon is shutting down+  initiateShutdown daemonShutdownLatch++  -- Stop the push manager and wait for any remaining paths to be uploaded+  shutdownPushManager daemonPushManager++  -- Stop worker threads+  withTakeMVar daemonWorkerThreads Worker.stopWorkers++  -- Close all event subscriptions+  withTakeMVar daemonSubscriptionManagerThread (shutdownSubscriptions daemonSubscriptionManager)++  failedJobs <-+    PushManager.runPushManager daemonPushManager PushManager.getFailedPushJobs+  let pushResult =+        if null failedJobs+          then Right ()+          else Left DaemonPushFailure++  -- Gracefully close open connections to clients+  Async.mapConcurrently_ (sayGoodbye pushResult) =<< SocketStore.toList daemonClients++  return pushResult+  where+    shutdownPushManager daemonPushManager = do+      queuedStorePathCount <- PushManager.runPushManager daemonPushManager PushManager.queuedStorePathCount+      when (queuedStorePathCount > 0) $+        Katip.logFM Katip.InfoS $+          Katip.logStr $+            "Remaining store paths: " <> (show queuedStorePathCount :: Text)++      Katip.logFM Katip.DebugS "Waiting for push manager to clear remaining jobs..."+      -- Finish processing remaining push jobs+      let timeoutOptions =+            PushManager.TimeoutOptions+              { PushManager.toTimeout = 60.0,+                PushManager.toPollingInterval = 1.0+              }+      liftIO $ PushManager.stopPushManager timeoutOptions daemonPushManager+      Katip.logFM Katip.DebugS "Push manager shut down."++    shutdownSubscriptions daemonSubscriptionManager subscriptionManagerThread = do+      Katip.logFM Katip.DebugS "Shutting down event manager..."+      liftIO $ stopSubscriptionManager daemonSubscriptionManager+      _ <- Async.wait subscriptionManagerThread+      Katip.logFM Katip.DebugS "Event manager shut down."++    sayGoodbye exitResult socket = do+      let clientSock = SocketStore.socket socket+      let clientThread = SocketStore.handlerThread socket+      Async.cancel clientThread++      -- Wave goodbye to the client that requested the shutdown+      liftIO $ Listen.serverBye clientSock exitResult+      liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` (\_ -> return ())+      -- Wait for the other end to disconnect+      ebs <- liftIO $ try $ Socket.BS.recv clientSock 4096+      case ebs of+        Left err | isResourceVanishedError err -> Katip.logFM Katip.DebugS "Client did not disconnect cleanly."+        Left err -> Katip.logFM Katip.DebugS $ Katip.ls $ "Client socket threw an error: " <> displayException err+        Right _ -> Katip.logFM Katip.DebugS "Client disconnected."++withTakeMVar :: (MonadUnliftIO m) => MVar a -> (a -> m ()) -> m ()+withTakeMVar mvar f = do+  bracket acquire release wrapper+  where+    acquire = liftIO $ tryTakeMVar mvar++    release Nothing = pure ()+    release (Just x) = liftIO $ putMVar mvar x++    wrapper Nothing = pure ()+    wrapper (Just x) = f x
+ src/Cachix/Daemon/Client.hs view
@@ -0,0 +1,144 @@+module Cachix.Daemon.Client (push, stop) where++import Cachix.Client.Env as Env+import Cachix.Client.OptionsParser (DaemonOptions (..))+import Cachix.Client.Retry qualified as Retry+import Cachix.Daemon.Listen (getSocketPath)+import Cachix.Daemon.Protocol as Protocol+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM.TBMQueue+import Data.Aeson qualified as Aeson+import Data.ByteString qualified as BS+import Data.IORef+import Data.Time.Clock+import Network.Socket qualified as Socket+import Network.Socket.ByteString qualified as Socket.BS+import Network.Socket.ByteString.Lazy qualified as Socket.LBS+import Protolude+import System.IO.Error (isResourceVanishedError)++data SocketError+  = -- | The socket has been closed+    SocketClosed+  | -- | The socket has stopped responding to pings+    SocketStalled+  | -- | Failed to decode a message from the socket+    SocketDecodingError !Text+  deriving stock (Show)++instance Exception SocketError where+  displayException = \case+    SocketClosed -> "The socket has been closed"+    SocketStalled -> "The socket has stopped responding to pings"+    SocketDecodingError err -> "Failed to decode the message from socket: " <> toS err++-- | Queue up push requests with the daemon+--+-- TODO: wait for the daemon to respond that it has received the request+push :: Env -> DaemonOptions -> [FilePath] -> IO ()+push _env daemonOptions storePaths =+  withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do+    Socket.LBS.sendAll sock $ Protocol.newMessage pushRequest+  where+    pushRequest =+      Protocol.ClientPushRequest $+        PushRequest {storePaths = storePaths}++-- | Tell the daemon to stop and wait for it to gracefully exit+stop :: Env -> DaemonOptions -> IO ()+stop _env daemonOptions =+  withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do+    let size = 100+    (rx, tx) <- atomically $ (,) <$> newTBMQueue size <*> newTBMQueue size++    rxThread <- Async.async (handleIncoming rx sock)+    txThread <- Async.async (handleOutgoing tx sock)++    lastPongRef <- newIORef =<< getCurrentTime+    pingThread <- Async.async (runPingThread lastPongRef rx tx)++    mapM_ Async.link [rxThread, txThread, pingThread]++    -- Request the daemon to stop+    atomically $ writeTBMQueue tx Protocol.ClientStop++    fix $ \loop -> do+      mmsg <- atomically (readTBMQueue rx)+      case mmsg of+        Nothing -> return ()+        Just (Left err) -> do+          putErrText $ toS $ displayException err+          exitFailure+        Just (Right msg) ->+          case msg of+            Protocol.DaemonPong -> do+              writeIORef lastPongRef =<< getCurrentTime+              loop+            Protocol.DaemonExit exitStatus ->+              case exitCode exitStatus of+                0 -> exitSuccess+                code -> exitWith (ExitFailure code)+  where+    runPingThread lastPongRef rx tx = go+      where+        go = do+          timestamp <- getCurrentTime+          lastPong <- readIORef lastPongRef++          if timestamp >= addUTCTime 20 lastPong+            then atomically $ writeTBMQueue rx (Left SocketStalled)+            else do+              atomically $ writeTBMQueue tx Protocol.ClientPing+              threadDelay (2 * 1000 * 1000)+              go++    handleOutgoing tx sock = go+      where+        go = do+          mmsg <- atomically $ readTBMQueue tx+          case mmsg of+            Nothing -> return ()+            Just msg -> do+              Retry.retryAll $ const $ Socket.LBS.sendAll sock $ Protocol.newMessage msg+              go++    handleIncoming rx sock = go BS.empty+      where+        socketClosed = atomically $ writeTBMQueue rx (Left SocketClosed)++        go leftovers = do+          ebs <- liftIO $ try $ Socket.BS.recv sock 4096++          case ebs of+            Left err | isResourceVanishedError err -> socketClosed+            Left _err -> socketClosed+            -- If the socket returns 0 bytes, then it is closed+            Right bs | BS.null bs -> socketClosed+            Right bs -> do+              let (rawMsgs, newLeftovers) = Protocol.splitMessages (BS.append leftovers bs)++              forM_ (map Aeson.eitherDecodeStrict rawMsgs) $ \emsg -> do+                case emsg of+                  Left err -> do+                    let terr = toS err+                    putErrText terr+                    atomically $ writeTBMQueue rx (Left (SocketDecodingError terr))+                  Right msg -> do+                    atomically $ writeTBMQueue rx (Right msg)++              go newLeftovers++withDaemonConn :: Maybe FilePath -> (Socket.Socket -> IO a) -> IO a+withDaemonConn optionalSocketPath f = do+  socketPath <- maybe getSocketPath pure optionalSocketPath+  bracket (open socketPath `onException` failedToConnectTo socketPath) Socket.close f+  where+    open socketPath = do+      sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol+      Retry.retryAll $ const $ Socket.connect sock (Socket.SockAddrUnix socketPath)+      return sock++    failedToConnectTo :: FilePath -> IO ()+    failedToConnectTo socketPath = do+      putErrText "\nFailed to connect to Cachix Daemon"+      putErrText $ "Tried to connect to: " <> toS socketPath <> "\n"
+ src/Cachix/Daemon/EventLoop.hs view
@@ -0,0 +1,78 @@+module Cachix.Daemon.EventLoop+  ( new,+    send,+    sendIO,+    run,+    exitLoopWith,+    EventLoop,+  )+where++import Cachix.Daemon.Types.EventLoop (EventLoop (..), EventLoopError (..))+import Control.Concurrent.STM.TBMQueue+  ( isFullTBMQueue,+    newTBMQueueIO,+    readTBMQueue,+    tryWriteTBMQueue,+  )+import Data.Text.Lazy.Builder (toLazyText)+import Katip qualified+import Protolude++new :: (MonadIO m) => m (EventLoop event a)+new = do+  exitLatch <- liftIO newEmptyMVar+  queue <- liftIO $ newTBMQueueIO 100+  return $ EventLoop {queue, exitLatch}++-- | Send an event to the event loop with logging.+send :: (Katip.KatipContext m) => EventLoop event a -> event -> m ()+send = send' Katip.logFM++-- | Same as 'send', but does not require a 'Katip.KatipContext'.+sendIO :: forall m event a. (MonadIO m) => EventLoop event a -> event -> m ()+sendIO = send' logger+  where+    logger :: Katip.Severity -> Katip.LogStr -> m ()+    logger Katip.ErrorS msg = liftIO $ hPutStrLn stderr (toLazyText $ Katip.unLogStr msg)+    logger _ _ = return ()++send' :: (MonadIO m) => (Katip.Severity -> Katip.LogStr -> m ()) -> EventLoop event a -> event -> m ()+send' logger eventloop@(EventLoop {queue}) event = do+  res <- liftIO $ atomically $ tryWriteTBMQueue queue event+  case res of+    -- The queue is closed.+    Nothing ->+      logger Katip.DebugS "Ignored an event because the event loop is closed"+    -- Successfully wrote to the queue+    Just True -> return ()+    -- Failed to write to the queue+    Just False -> do+      isFull <- liftIO $ atomically $ isFullTBMQueue queue+      let message =+            if isFull+              then "Event loop is full"+              else "Unknown error"+      logger Katip.ErrorS $ "Failed to write to event loop: " <> message+      exitLoopWithFailure EventLoopFull eventloop++-- | Run the event loop until it exits with 'exitLoopWith'.+run :: (MonadIO m) => EventLoop event a -> (event -> m ()) -> m (Either EventLoopError a)+run eventloop f = do+  fix $ \loop -> do+    mevent <- liftIO $ atomically $ readTBMQueue (queue eventloop)+    case mevent of+      Just event -> f event+      Nothing -> exitLoopWithFailure EventLoopClosed eventloop++    liftIO (tryReadMVar (exitLatch eventloop)) >>= \case+      Just exitValue -> return exitValue+      Nothing -> loop++-- | Short-circuit the event loop and exit with a given return value.+exitLoopWith :: (MonadIO m) => a -> EventLoop event a -> m ()+exitLoopWith exitValue (EventLoop {exitLatch}) = void $ liftIO $ tryPutMVar exitLatch (Right exitValue)++-- | Short-circuit the event loop in case of an internal error.+exitLoopWithFailure :: (MonadIO m) => EventLoopError -> EventLoop event a -> m ()+exitLoopWithFailure err (EventLoop {exitLatch}) = void $ liftIO $ tryPutMVar exitLatch (Left err)
+ src/Cachix/Daemon/Listen.hs view
@@ -0,0 +1,162 @@+module Cachix.Daemon.Listen+  ( listen,+    handleClient,+    serverBye,+    getSocketPath,+    openSocket,+    closeSocket,+  )+where++import Cachix.Client.Config.Orphans ()+import Cachix.Daemon.EventLoop qualified as EventLoop+import Cachix.Daemon.Protocol as Protocol+import Cachix.Daemon.Types+  ( DaemonError,+    DaemonEvent+      ( AddSocketClient,+        ReceivedMessage,+        RemoveSocketClient+      ),+    toExitCodeInt,+  )+import Cachix.Daemon.Types.EventLoop (EventLoop)+import Cachix.Daemon.Types.SocketStore (SocketId)+import Control.Exception.Safe (catchAny)+import Control.Monad.Catch qualified as E+import Data.Aeson qualified as Aeson+import Data.ByteString qualified as BS+import Katip qualified+import Network.Socket (Socket)+import Network.Socket qualified as Socket+import Network.Socket.ByteString qualified as Socket.BS+import Network.Socket.ByteString.Lazy qualified as Socket.LBS+import Protolude+import System.Directory+  ( XdgDirectory (..),+    createDirectoryIfMissing,+    getXdgDirectory,+    removeFile,+  )+import System.Environment qualified as System+import System.FilePath ((</>))+import System.IO.Error (isDoesNotExistError, isResourceVanishedError)++-- TODO: reconcile with Client+data ListenError+  = SocketError SomeException+  | DecodingError Text+  deriving stock (Show)++instance Exception ListenError where+  displayException = \case+    SocketError err -> "Failed to read from the daemon socket: " <> show err+    DecodingError err -> "Failed to decode request:\n" <> toS err++-- | Listen for incoming connections on the given socket path.+listen ::+  (E.MonadMask m, Katip.KatipContext m) =>+  EventLoop DaemonEvent a ->+  FilePath ->+  m ()+listen eventloop daemonSocketPath = forever $ do+  E.bracketOnError (openSocket daemonSocketPath) closeSocket $ \sock -> do+    liftIO $ Socket.listen sock Socket.maxListenQueue+    (conn, _peerAddr) <- liftIO $ Socket.accept sock+    EventLoop.send eventloop (AddSocketClient conn)++-- | Handle incoming messages from a client.+--+-- Automatically responds to pings.+-- Requests the daemon to remove the client socket once the loop exits.+handleClient ::+  forall m a.+  (E.MonadMask m, Katip.KatipContext m) =>+  EventLoop DaemonEvent a ->+  SocketId ->+  Socket ->+  m ()+handleClient eventloop socketId conn = do+  go BS.empty `E.finally` removeClient+  where+    go leftovers = do+      ebs <- liftIO $ try $ Socket.BS.recv conn 8192++      case ebs of+        Left err | isResourceVanishedError err -> return ()+        Left _ -> return ()+        -- If the socket returns 0 bytes, then it is closed+        Right bs | BS.null bs -> return ()+        Right bs -> do+          let (rawMsgs, newLeftovers) = Protocol.splitMessages (BS.append leftovers bs)+          msgs <- catMaybes <$> mapM decodeMessage rawMsgs++          forM_ msgs $ \msg -> do+            EventLoop.send eventloop (ReceivedMessage msg)+            case msg of+              Protocol.ClientPing ->+                liftIO $ Socket.LBS.sendAll conn $ Protocol.newMessage DaemonPong+              _ -> return ()++          go newLeftovers++    removeClient = EventLoop.send eventloop (RemoveSocketClient socketId)++decodeMessage :: (Katip.KatipContext m) => ByteString -> m (Maybe Protocol.ClientMessage)+decodeMessage "" = return Nothing+decodeMessage bs =+  case Aeson.eitherDecodeStrict bs of+    Left err -> do+      Katip.logFM Katip.ErrorS $ Katip.ls $ displayException (DecodingError (toS err))+      return Nothing+    Right msg -> return (Just msg)++serverBye :: Socket.Socket -> Either DaemonError () -> IO ()+serverBye sock exitResult =+  Socket.LBS.sendAll sock (Protocol.newMessage (DaemonExit exitStatus)) `catchAny` (\_ -> return ())+  where+    exitStatus = DaemonExitStatus {exitCode, exitMessage}+    exitCode = toExitCodeInt exitResult+    exitMessage =+      case exitResult of+        Left err -> Just (toS $ displayException err)+        Right _ -> Nothing++getSocketPath :: IO FilePath+getSocketPath = do+  socketDir <- getSocketDir+  return $ socketDir </> "cachix-daemon.sock"++getSocketDir :: IO FilePath+getSocketDir = do+  xdgRuntimeDir <- getXdgRuntimeDir+  let socketDir = xdgRuntimeDir </> "cachix"+  createDirectoryIfMissing True socketDir+  return socketDir++-- On systems with systemd: /run/user/$UID+-- Otherwise, fall back to XDG_CACHE_HOME+getXdgRuntimeDir :: IO FilePath+getXdgRuntimeDir = do+  xdgRuntimeDir <- System.lookupEnv "XDG_RUNTIME_DIR"+  cacheFallback <- getXdgDirectory XdgCache ""+  return $ fromMaybe cacheFallback xdgRuntimeDir++-- TODO: lock the socket+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+  -- setFileMode socketFilePath socketFileMode+  return sock+  where+    deleteSocketFileIfExists path =+      removeFile path `catch` handleDoesNotExist++    handleDoesNotExist e+      | isDoesNotExistError e = return ()+      | otherwise = throwIO e++closeSocket :: (MonadIO m) => Socket.Socket -> m ()+closeSocket = liftIO . Socket.close
+ src/Cachix/Daemon/Log.hs view
@@ -0,0 +1,79 @@+module Cachix.Daemon.Log+  ( new,+    withLogger,+    getKatipNamespace,+    getKatipContext,+    getKatipLogEnv,+    localLogEnv,+    localKatipContext,+    localKatipNamespace,+    toKatipLogLevel,+    Log.Logger (..),+    Log.LogLevel (..),+  )+where++import Cachix.Daemon.Types.Log as Log+import Control.Monad.Catch qualified as E+import Data.Text.Lazy.Builder+import Katip (renderSeverity)+import Katip qualified+import Katip.Format.Time qualified as Katip.Format+import Katip.Scribes.Handle (brackets, colorBySeverity, getKeys)+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))+    <> mconcat ks+    <> fromText " "+    <> Katip.unLogStr _itemMessage+  where+    nowStr = fromText (Katip.Format.formatAsLogTime _itemTime)+    ks = map brackets $ getKeys verbosity _itemPayload+    renderSeverity' severity =+      colorBySeverity withColor severity (renderSeverity severity)
+ src/Cachix/Daemon/PostBuildHook.hs view
@@ -0,0 +1,154 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TypeApplications #-}++module Cachix.Daemon.PostBuildHook+  ( -- * Post-build hook setup+    PostBuildHookEnv (..),+    withSetup,++    -- * Set up env vars+    EnvVar,+    modifyEnv,++    -- * Internal+    buildNixConfEnv,+    buildNixUserConfFilesEnv,+  )+where++import Control.Monad.Catch (MonadMask)+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 (getCanonicalTemporaryDirectory, withTempDirectory)+import System.Posix.Files++type EnvVar = (String, String)++modifyEnv :: EnvVar -> [EnvVar] -> [EnvVar]+modifyEnv (envName, envValue) processEnv =+  nubOrd $ (envName, envValue) : processEnv++data PostBuildHookEnv = PostBuildHookEnv+  { tempDir :: !FilePath,+    postBuildHookConfigPath :: !FilePath,+    postBuildHookScriptPath :: !FilePath,+    daemonSock :: !FilePath,+    envVar :: !EnvVar+  }++withSetup :: Maybe FilePath -> (PostBuildHookEnv -> IO a) -> IO a+withSetup mdaemonSock f =+  withRunnerFriendlyTempDirectory "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+    envVar <- case mnixConfEnv of+      Just nixConfEnv -> return nixConfEnv+      Nothing -> do+        writeFile postBuildHookConfigPath (postBuildHookConfig postBuildHookScriptPath)+        return nixUserConfFilesEnv++    f PostBuildHookEnv {..}++-- | 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+  |]++-- | Run an action with a temporary directory.+--+-- Respects the RUNNER_TEMP environment variable if set.+-- This is important on self-hosted GitHub runners with locked down system temp directories.+-- The directory is deleted after use.+withRunnerFriendlyTempDirectory :: (MonadIO m, MonadMask m) => String -> (FilePath -> m a) -> m a+withRunnerFriendlyTempDirectory name action = do+  runnerTempDir <- liftIO $ lookupEnv "RUNNER_TEMP"+  systemTempDir <- liftIO getCanonicalTemporaryDirectory+  let tempDir = maybe systemTempDir toS runnerTempDir+  withTempDirectory tempDir name action
+ src/Cachix/Daemon/Progress.hs view
@@ -0,0 +1,168 @@+module Cachix.Daemon.Progress+  ( UploadProgress,+    new,+    complete,+    fail,+    tick,+    update,+  )+where++import Cachix.Client.HumanSize (humanSize)+import Cachix.Daemon.Types (PushRetryStatus (..))+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar+import Data.String (String)+import Data.Text qualified as T+import Protolude+import System.Console.AsciiProgress qualified as Ascii+import System.Console.AsciiProgress.Internal qualified as Ascii.Internal+import System.Console.Pretty+import System.Environment (lookupEnv)+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+  isCI <- liftIO $ (== Just "true") <$> lookupEnv "CI"+  isTerminal <- liftIO $ hIsTerminalDevice hdl+  if isTerminal && not isCI+    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/Daemon/Protocol.hs view
@@ -0,0 +1,75 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Cachix.Daemon.Protocol+  ( ClientMessage (..),+    DaemonMessage (..),+    DaemonExitStatus (..),+    PushRequestId,+    newPushRequestId,+    PushRequest (..),+    newMessage,+    splitMessages,+  )+where++import Data.Aeson qualified as Aeson+import Data.ByteString qualified as BS+import Data.ByteString.Lazy qualified as LBS+import Data.UUID (UUID)+import Data.UUID.V4 qualified as UUID+import Protolude++-- | JSON messages that the client can send to the daemon+data ClientMessage+  = ClientPushRequest !PushRequest+  | ClientStop+  | ClientPing+  deriving stock (Eq, Generic, Show)+  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)++-- | JSON messages that the daemon can send to the client+data DaemonMessage+  = DaemonPong+  | DaemonExit !DaemonExitStatus+  deriving stock (Eq, Generic, Show)+  deriving anyclass (Aeson.FromJSON, Aeson.ToJSON)++data DaemonExitStatus = DaemonExitStatus+  { exitCode :: !Int,+    exitMessage :: !(Maybe Text)+  }+  deriving stock (Eq, Generic, Show)+  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)++newMessage :: (Aeson.ToJSON a) => a -> LBS.ByteString+newMessage msg =+  Aeson.encode msg `LBS.snoc` fromIntegral (fromEnum '\n')++splitMessages :: ByteString -> ([ByteString], ByteString)+splitMessages = go []+  where+    go acc bs =+      case BS.break (== _n) bs of+        (line, "") -> (reverse acc, line)+        (line, "\n") -> go (line `cons` acc) mempty+        (line, rest) ->+          go (line `cons` acc) (BS.drop 1 rest)++    cons "" xs = xs+    cons x xs = x : xs++    _n = fromIntegral (fromEnum '\n')
+ src/Cachix/Daemon/Push.hs view
@@ -0,0 +1,68 @@+module Cachix.Daemon.Push+  ( newPushParams,+    getBinaryCache,+    getCompressionMethod,+  )+where++import Cachix.API qualified as API+import Cachix.Client.Command.Push hiding (pushStrategy)+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.Daemon.PushManager qualified as PushManager+import Cachix.Daemon.Types (PushManager)+import Cachix.Types.BinaryCache (BinaryCache, BinaryCacheName)+import Cachix.Types.BinaryCache qualified as BinaryCache+import Data.Set qualified as Set+import Hercules.CNix.Store (Store)+import Protolude hiding (toS)+import Servant.Auth ()+import Servant.Auth.Client+import Servant.Client.Streaming+import Servant.Conduit ()++newPushParams ::+  Store ->+  ClientEnv ->+  BinaryCache ->+  PushSecret ->+  PushOptions ->+  PushParams PushManager ()+newPushParams store clientEnv binaryCache pushSecret pushOptions = do+  let authToken = getAuthTokenFromPushSecret pushSecret+      cacheName = BinaryCache.name binaryCache+      compressionMethod = getCompressionMethod pushOptions binaryCache+      pushStrategy = PushManager.newPushStrategy store authToken pushOptions cacheName compressionMethod++  PushParams+    { pushParamsName = cacheName,+      pushParamsSecret = pushSecret,+      pushParamsClientEnv = clientEnv,+      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+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.CompressionMethod+getCompressionMethod opts binaryCache =+  fromMaybe BinaryCache.ZSTD $+    Client.OptionsParser.compressionMethod opts+      <|> Just (BinaryCache.preferredCompressionMethod binaryCache)
+ src/Cachix/Daemon/PushManager.hs view
@@ -0,0 +1,516 @@+module Cachix.Daemon.PushManager+  ( newPushManagerEnv,+    runPushManager,+    stopPushManager,++    -- * Push strategy+    newPushStrategy,++    -- * Push job+    PushJob (..),+    addPushJob,+    lookupPushJob,+    withPushJob,+    resolvePushJob,+    pendingJobCount,++    -- * Query+    filterPushJobs,+    getFailedPushJobs,++    -- * Store paths+    queueStorePaths,+    removeStorePath,+    queuedStorePathCount,++    -- * Tasks+    handleTask,++    -- * Push events+    pushStarted,+    pushFinished,+    pushStorePathAttempt,+    pushStorePathProgress,+    pushStorePathDone,+    pushStorePathFailed,++    -- * Helpers+    atomicallyWithTimeout,+  )+where++import Cachix.Client.CNix (filterInvalidStorePath, followLinksToStorePath)+import Cachix.Client.Command.Push hiding (pushStrategy)+import Cachix.Client.OptionsParser as Client.OptionsParser+  ( PushOptions (..),+  )+import Cachix.Client.Push as Client.Push+import Cachix.Client.Retry (retryAll)+import Cachix.Daemon.Protocol qualified as Protocol+import Cachix.Daemon.PushManager.PushJob qualified as PushJob+import Cachix.Daemon.Types.Log (Logger)+import Cachix.Daemon.Types.PushEvent+import Cachix.Daemon.Types.PushManager+import Cachix.Types.BinaryCache qualified as BinaryCache+import Conduit qualified as C+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM.TBMQueue+import Control.Concurrent.STM.TVar+import Control.Monad.Catch qualified as E+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import Control.Retry (RetryStatus)+import Data.ByteString qualified as BS+import Data.HashMap.Strict qualified as HashMap+import Data.IORef+import Data.Set qualified as Set+import Data.Text qualified as T+import Data.Time (UTCTime, diffUTCTime, getCurrentTime, secondsToNominalDiffTime)+import Hercules.CNix (StorePath)+import Hercules.CNix.Store (Store, parseStorePath, storePathToPath)+import Katip qualified+import Protolude hiding (toS)+import Protolude.Conv (toS)+import Servant.Auth ()+import Servant.Auth.Client+import Servant.Conduit ()+import UnliftIO.QSem qualified as QSem++newPushManagerEnv :: (MonadIO m) => PushOptions -> PushParams PushManager () -> OnPushEvent -> Logger -> m PushManagerEnv+newPushManagerEnv pushOptions pmPushParams onPushEvent pmLogger = liftIO $ do+  pmPushJobs <- newTVarIO mempty+  pmPendingJobCount <- newTVarIO 0+  pmStorePathIndex <- newTVarIO mempty+  pmTaskQueue <- newTBMQueueIO 1000+  pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)+  pmLastEventTimestamp <- newTVarIO =<< getCurrentTime+  let pmOnPushEvent id pushEvent = updateTimestampTVar pmLastEventTimestamp >> onPushEvent id pushEvent++  return $ PushManagerEnv {..}++runPushManager :: (MonadIO m) => PushManagerEnv -> PushManager a -> m a+runPushManager env f = liftIO $ unPushManager f `runReaderT` env++stopPushManager :: TimeoutOptions -> PushManagerEnv -> IO ()+stopPushManager timeoutOptions PushManagerEnv {..} = do+  atomicallyWithTimeout timeoutOptions pmLastEventTimestamp $ do+    pendingJobs <- readTVar pmPendingJobCount+    check (pendingJobs <= 0)+  atomically $ closeTBMQueue pmTaskQueue++-- Manage push jobs++addPushJob :: Protocol.PushRequest -> PushManager (Maybe Protocol.PushRequestId)+addPushJob pushRequest = do+  PushManagerEnv {..} <- ask+  pushJob <- PushJob.new pushRequest++  Katip.logLocM Katip.DebugS $ Katip.ls $ "Queued push job " <> (show (pushId pushJob) :: Text)++  let queueJob = do+        modifyTVar' pmPushJobs $ HashMap.insert (pushId pushJob) pushJob+        incrementTVar pmPendingJobCount+        res <- tryWriteTBMQueue pmTaskQueue $ ResolveClosure (pushId pushJob)+        case res of+          Just True -> return True+          _ -> retry++  didQueue <- liftIO $ atomically $ queueJob `orElse` return False++  if didQueue+    then return $ Just $ pushId pushJob+    else do+      Katip.logLocM Katip.WarningS "Failed to queue push job. Queue likely full."+      return Nothing++removePushJob :: Protocol.PushRequestId -> PushManager ()+removePushJob pushId = do+  PushManagerEnv {..} <- ask+  liftIO $ atomically $ do+    mpushJob <- HashMap.lookup pushId <$> readTVar pmPushJobs+    for_ mpushJob $ \pushJob -> do+      -- Decrement the job count if this job had not been processed yet+      unless (PushJob.isProcessed pushJob) (decrementTVar pmPendingJobCount)+      modifyTVar' pmPushJobs (HashMap.delete pushId)++lookupPushJob :: Protocol.PushRequestId -> PushManager (Maybe PushJob)+lookupPushJob pushId = do+  pushJobs <- asks pmPushJobs+  liftIO $ HashMap.lookup pushId <$> readTVarIO pushJobs++filterPushJobs :: (PushJob -> Bool) -> PushManager [PushJob]+filterPushJobs f = do+  pushJobs <- asks pmPushJobs+  liftIO $ filter f . HashMap.elems <$> readTVarIO pushJobs++getFailedPushJobs :: PushManager [PushJob]+getFailedPushJobs = filterPushJobs PushJob.isFailed++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 $ modifyPushJobSTM pushJobs pushId f++modifyPushJobSTM :: PushJobStore -> Protocol.PushRequestId -> (PushJob -> PushJob) -> STM (Maybe PushJob)+modifyPushJobSTM pushJobs pushId f =+  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++failPushJob :: Protocol.PushRequestId -> PushManager ()+failPushJob pushId = do+  PushManagerEnv {..} <- ask+  timestamp <- liftIO getCurrentTime+  liftIO $ atomically $ do+    _ <- modifyPushJobSTM pmPushJobs pushId $ PushJob.fail timestamp+    decrementTVar pmPendingJobCount++pendingJobCount :: PushManager Int+pendingJobCount = do+  pmPendingJobCount <- asks pmPendingJobCount+  liftIO $ readTVarIO pmPendingJobCount++-- Manage store paths++queueStorePaths :: Protocol.PushRequestId -> [FilePath] -> PushManager ()+queueStorePaths pushId storePaths = do+  PushManagerEnv {..} <- ask++  let addToQueue storePath = do+        isDuplicate <- HashMap.member storePath <$> readTVar pmStorePathIndex+        unless isDuplicate $+          writeTBMQueue pmTaskQueue (PushStorePath storePath)++        modifyTVar' pmStorePathIndex $ HashMap.insertWith (<>) storePath [pushId]++  transactionally $ map addToQueue storePaths++removeStorePath :: FilePath -> PushManager ()+removeStorePath storePath = do+  storePathIndex <- asks pmStorePathIndex+  liftIO $ atomically $ do+    modifyTVar' storePathIndex $ HashMap.delete storePath++lookupStorePathIndex :: FilePath -> PushManager [Protocol.PushRequestId]+lookupStorePathIndex storePath = do+  storePathIndex <- asks pmStorePathIndex+  references <- liftIO $ readTVarIO storePathIndex+  return $ fromMaybe [] (HashMap.lookup storePath references)++checkPushJobCompleted :: Protocol.PushRequestId -> PushManager ()+checkPushJobCompleted pushId = do+  PushManagerEnv {..} <- ask+  mpushJob <- lookupPushJob pushId+  for_ mpushJob $ \pushJob ->+    when (Set.null $ pushQueue pushJob) $ do+      timestamp <- liftIO getCurrentTime+      liftIO $ atomically $ do+        _ <- modifyPushJobSTM pmPushJobs pushId $ \pushJob' ->+          if PushJob.hasFailedPaths pushJob'+            then PushJob.fail timestamp pushJob'+            else PushJob.complete timestamp pushJob'+        decrementTVar pmPendingJobCount+      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+  Katip.logLocM Katip.DebugS $ Katip.ls $ showClosureStats closure++  timestamp <- liftIO getCurrentTime+  _ <- modifyPushJob pushId $ PushJob.populateQueue closure timestamp++  withPushJob pushId $ \pushJob -> do+    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 :: Task -> PushManager ()+handleTask task = do+  pushParams <- asks pmPushParams+  case task of+    ResolveClosure pushId ->+      runResolveClosureTask pushParams pushId+    PushStorePath filePath ->+      runPushStorePathTask pushParams filePath++runResolveClosureTask :: PushParams PushManager () -> Protocol.PushRequestId -> PushManager ()+runResolveClosureTask pushParams pushId =+  resolveClosure `withException` failJob+  where+    failJob :: SomeException -> PushManager ()+    failJob err = do+      failPushJob pushId++      Katip.katipAddContext (Katip.sl "error" (displayException err)) $+        Katip.logLocM Katip.ErrorS $+          Katip.ls $+            "Failed to resolve closure for push job " <> (show pushId :: Text)++    resolveClosure = do+      Katip.logLocM Katip.DebugS $ Katip.ls $ "Resolving closure for push job " <> (show pushId :: Text)++      withPushJob pushId $ \pushJob -> 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++runPushStorePathTask :: PushParams PushManager () -> FilePath -> PushManager ()+runPushStorePathTask pushParams filePath = do+  pushStorePath `withException` failStorePath+  where+    failStorePath =+      pushStorePathFailed filePath . toS . displayException++    pushStorePath = do+      qs <- asks pmTaskSemaphore+      E.bracket_ (QSem.waitQSem qs) (QSem.signalQSem qs) $ do+        Katip.logLocM Katip.DebugS $ Katip.ls $ "Pushing store path " <> filePath++        let store = pushParamsStore pushParams+        storePath <- liftIO $ parseStorePath store (toS filePath)++        retryAll $ uploadStorePath pushParams storePath++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.chunkSize = Client.OptionsParser.chunkSize opts,+          Client.Push.numConcurrentChunks = Client.OptionsParser.numConcurrentChunks 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 <- lookupStorePathIndex storePath+  sendStorePathEvent pushIds (PushStorePathAttempt storePath size pushRetryStatus)++pushStorePathProgress :: FilePath -> Int64 -> Int64 -> PushManager ()+pushStorePathProgress storePath currentBytes newBytes = do+  pushIds <- lookupStorePathIndex storePath+  sendStorePathEvent pushIds (PushStorePathProgress storePath currentBytes newBytes)++pushStorePathDone :: FilePath -> PushManager ()+pushStorePathDone storePath = do+  pushIds <- lookupStorePathIndex storePath+  modifyPushJobs pushIds (PushJob.markStorePathPushed storePath)++  sendStorePathEvent pushIds (PushStorePathDone storePath)++  mapM_ checkPushJobCompleted pushIds++  removeStorePath storePath++pushStorePathFailed :: FilePath -> Text -> PushManager ()+pushStorePathFailed storePath errMsg = do+  pushIds <- lookupStorePathIndex storePath+  modifyPushJobs pushIds (PushJob.markStorePathFailed storePath)++  sendStorePathEvent pushIds (PushStorePathFailed storePath errMsg)++  mapM_ checkPushJobCompleted pushIds++  removeStorePath storePath++-- Helpers++storeToFilePath :: (MonadIO m) => Store -> StorePath -> m FilePath+storeToFilePath store storePath = do+  fp <- liftIO $ storePathToPath store storePath+  pure $ toS fp++-- | Canonicalize and validate a store path+normalizeStorePath :: (MonadIO m) => Store -> FilePath -> m (Maybe StorePath)+normalizeStorePath store fp =+  liftIO $ runMaybeT $ do+    storePath <- MaybeT $ followLinksToStorePath store (encodeUtf8 $ T.pack fp)+    MaybeT $ filterInvalidStorePath store storePath++withException :: (E.MonadCatch m) => m a -> (SomeException -> m a) -> m a+withException action handler = action `E.catchAll` (\e -> handler e >> E.throwM e)++-- STM helpers++transactionally :: (Foldable t, MonadIO m) => t (STM ()) -> m ()+transactionally = liftIO . atomically . sequence_++updateTimestampTVar :: (MonadIO m) => TVar UTCTime -> m ()+updateTimestampTVar tvar = liftIO $ do+  now <- getCurrentTime+  atomically $ writeTVar tvar now++incrementTVar :: TVar Int -> STM ()+incrementTVar tvar = modifyTVar' tvar (+ 1)++decrementTVar :: TVar Int -> STM ()+decrementTVar tvar = modifyTVar' tvar (subtract 1)++-- | Run a transaction with a timeout.+atomicallyWithTimeout ::+  TimeoutOptions ->+  -- | A TVar timestamp to compare against+  TVar UTCTime ->+  -- | The transaction to run+  STM () ->+  IO ()+atomicallyWithTimeout TimeoutOptions {..} timeVar transaction = do+  timeoutVar <- newTVarIO False+  Async.race_+    (updateShutdownTimeout timeoutVar)+    (waitForGracefulShutdown timeoutVar)+  where+    waitForGracefulShutdown timeout =+      atomically $ transaction `orElse` checkShutdownTimeout timeout++    updateShutdownTimeout timeoutVar =+      forever $ do+        now <- getCurrentTime+        atomically $ do+          timestamp <- readTVar timeVar+          let isTimeout =+                secondsToNominalDiffTime (realToFrac toTimeout) <= now `diffUTCTime` timestamp+          writeTVar timeoutVar isTimeout+        threadDelay $ ceiling (toPollingInterval * 1000.0 * 1000.0)++    checkShutdownTimeout timeout = check =<< readTVar timeout
+ src/Cachix/Daemon/PushManager/PushJob.hs view
@@ -0,0 +1,123 @@+module Cachix.Daemon.PushManager.PushJob+  ( module Cachix.Daemon.PushManager.PushJob,+    module Types,+  )+where++import Cachix.Daemon.Protocol qualified as Protocol+import Cachix.Daemon.Types.PushManager as Types+import Data.Set qualified 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+    { pushQueue = Set.delete storePath pushQueue,+      pushResult = addPushedPath storePath pushResult+    }++markStorePathFailed :: FilePath -> PushJob -> PushJob+markStorePathFailed storePath pushJob@(PushJob {pushQueue, pushResult}) =+  pushJob+    { 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++hasFailedPaths :: PushJob -> Bool+hasFailedPaths = not . Set.null . prFailedPaths . result++complete :: UTCTime -> PushJob -> PushJob+complete timestamp pushJob@PushJob {..} = do+  pushJob+    { pushStatus =+        case pushStatus of+          Running -> Completed+          _ -> pushStatus,+      pushStats = pushStats {jsCompletedAt = Just timestamp}+    }++fail :: UTCTime -> PushJob -> PushJob+fail timestamp pushJob@PushJob {..} = do+  pushJob+    { pushStatus = Failed,+      pushStats = pushStats {jsCompletedAt = Just timestamp}+    }++isCompleted :: PushJob -> Bool+isCompleted PushJob {pushStatus} = pushStatus == Completed++isFailed :: PushJob -> Bool+isFailed PushJob {pushStatus} = pushStatus == Failed++isProcessed :: PushJob -> Bool+isProcessed pushJob = isCompleted pushJob || isFailed pushJob++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/Daemon/ShutdownLatch.hs view
@@ -0,0 +1,26 @@+module Cachix.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/Daemon/SocketStore.hs view
@@ -0,0 +1,43 @@+module Cachix.Daemon.SocketStore+  ( newSocketStore,+    addSocket,+    removeSocket,+    toList,+    Socket (..),+  )+where++import Cachix.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..))+import Control.Concurrent.STM.TVar+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Data.HashMap.Strict qualified as HashMap+import Data.UUID.V4 qualified as UUID+import Network.Socket qualified+import Protolude hiding (toList)+import UnliftIO.Async qualified as Async++newSocketStore :: (MonadIO m) => m SocketStore+newSocketStore = SocketStore <$> liftIO (newTVarIO mempty)++newSocketId :: (MonadIO m) => m SocketId+newSocketId = liftIO UUID.nextRandom++addSocket :: (MonadUnliftIO m) => Network.Socket.Socket -> (SocketId -> Network.Socket.Socket -> m ()) -> SocketStore -> m ()+addSocket socket handler (SocketStore st) = do+  socketId <- newSocketId+  handlerThread <- Async.async (handler socketId socket)+  liftIO $ atomically $ modifyTVar' st $ HashMap.insert socketId (Socket {..})++removeSocket :: (MonadIO m) => SocketId -> SocketStore -> m ()+removeSocket socketId (SocketStore stvar) = do+  msocket <- liftIO $ atomically $ stateTVar stvar $ \st ->+    let msocket = HashMap.lookup socketId st+     in (msocket, HashMap.delete socketId st)++  -- shut down the handler thread+  mapM_ (Async.uninterruptibleCancel . handlerThread) msocket++toList :: (MonadIO m) => SocketStore -> m [Socket]+toList (SocketStore st) = do+  hm <- liftIO $ readTVarIO st+  return $ HashMap.elems hm
+ src/Cachix/Daemon/Subscription.hs view
@@ -0,0 +1,102 @@+module Cachix.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 Data.HashMap.Strict qualified as HashMap+import Network.Socket qualified 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/Daemon/Types.hs view
@@ -0,0 +1,32 @@+module Cachix.Daemon.Types+  ( -- * Daemon+    DaemonEvent (..),+    DaemonEnv (..),+    Daemon,+    runDaemon,++    -- * Daemon errors+    DaemonError (..),+    HasExitCode (..),++    -- * Log+    LogLevel (..),++    -- * Push+    PushManager.PushManagerEnv (..),+    PushManager.PushManager,+    PushManager.PushJob (..),+    Task (..),++    -- * Push Event+    PushEvent.PushEvent (..),+    PushEvent.PushEventMessage (..),+    PushEvent.PushRetryStatus (..),+  )+where++import Cachix.Daemon.Types.Daemon+import Cachix.Daemon.Types.Error+import Cachix.Daemon.Types.Log+import Cachix.Daemon.Types.PushEvent as PushEvent+import Cachix.Daemon.Types.PushManager as PushManager
+ src/Cachix/Daemon/Types/Daemon.hs view
@@ -0,0 +1,124 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TypeApplications #-}++module Cachix.Daemon.Types.Daemon+  ( -- * Daemon+    DaemonEvent (..),+    DaemonEnv (..),+    Daemon,+    runDaemon,+  )+where++import Cachix.Client.Config.Orphans ()+import Cachix.Client.Env as Env+import Cachix.Client.OptionsParser (PushOptions)+import Cachix.Daemon.Log qualified as Log+import Cachix.Daemon.Protocol qualified as Protocol+import Cachix.Daemon.ShutdownLatch (ShutdownLatch)+import Cachix.Daemon.Subscription (SubscriptionManager)+import Cachix.Daemon.Types.Error (DaemonError (DaemonUnhandledException), UnhandledException (..))+import Cachix.Daemon.Types.EventLoop (EventLoop)+import Cachix.Daemon.Types.Log (Logger)+import Cachix.Daemon.Types.PushEvent (PushEvent)+import Cachix.Daemon.Types.PushManager (PushManagerEnv (..))+import Cachix.Daemon.Types.SocketStore (SocketId, SocketStore)+import Cachix.Daemon.Worker qualified as Worker+import Cachix.Types.BinaryCache (BinaryCache, BinaryCacheName)+import Control.Exception.Safe qualified as Safe+import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow)+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Katip qualified+import Network.Socket (Socket)+import Protolude hiding (bracketOnError)+import System.Posix.Types (ProcessID)++-- | Daemon events that are handled by the 'EventLoop'.+data DaemonEvent+  = -- | Shut down the daemon gracefully.+    ShutdownGracefully+  | -- | Re-establish the daemon socket+    ReconnectSocket+  | -- | Add a new client socket connection.+    AddSocketClient Socket+  | -- | Remove an existing client socket connection. For example, after it is closed.+    RemoveSocketClient SocketId+  | -- | Handle a new message from a client.+    ReceivedMessage Protocol.ClientMessage++data DaemonEnv = DaemonEnv+  { -- | Cachix client env+    daemonEnv :: Env,+    -- | The main event loop+    daemonEventLoop :: EventLoop DaemonEvent (Either DaemonError ()),+    -- | Push options, like compression settings and number of jobs+    daemonPushOptions :: PushOptions,+    -- | Path to the socket that the daemon listens on+    daemonSocketPath :: FilePath,+    -- | Main inbound socket thread+    daemonSocketThread :: MVar (Async ()),+    -- | 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,+    -- | Connected clients over the socket+    daemonClients :: SocketStore,+    -- | A pool of worker threads+    daemonWorkerThreads :: MVar [Worker.Thread],+    -- | A multiplexer for push events+    daemonSubscriptionManager :: SubscriptionManager Protocol.PushRequestId PushEvent,+    -- | A thread handle for the subscription manager+    daemonSubscriptionManagerThread :: MVar (Async ()),+    -- | 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 (Either DaemonError a)+runDaemon env f = tryDaemon $ do+  Log.withLogger (daemonLogger env) $ \logger -> do+    let pushManagerEnv = (daemonPushManager env) {pmLogger = logger}+    unDaemon f `runReaderT` env {daemonLogger = logger, daemonPushManager = pushManagerEnv}+  where+    tryDaemon :: IO a -> IO (Either DaemonError a)+    tryDaemon a =+      Safe.catch+        (a >>= \v -> return (Right v))+        (return . Left . wrapUnhandledErrors)++    -- Wrap errors in a DaemonError+    wrapUnhandledErrors :: SomeException -> DaemonError+    wrapUnhandledErrors e | Just daemonError <- fromException @DaemonError e = daemonError+    wrapUnhandledErrors e = DaemonUnhandledException (UnhandledException e)
+ src/Cachix/Daemon/Types/Error.hs view
@@ -0,0 +1,65 @@+module Cachix.Daemon.Types.Error+  ( DaemonError (..),+    UnhandledException (..),+    HasExitCode (..),+  )+where++import Cachix.Daemon.Types.EventLoop (EventLoopError (..))+import Protolude++-- | An error that can occur in the daemon.+--+-- These should not escape the main 'run' function.+data DaemonError+  = -- | The daemon shut down due to an unhandled exception+    DaemonUnhandledException UnhandledException+  | -- | There was an error in the daemon's event loop+    DaemonEventLoopError EventLoopError+  | -- | There was an error with the daemon's socket+    DaemonSocketError+  | -- | Failed to push some store paths+    DaemonPushFailure+  deriving stock (Show, Eq)++instance Exception DaemonError where+  displayException =+    toS . \case+      DaemonUnhandledException e ->+        unlines ["Unhandled exception in the Cachix Daemon:", "", show e]+      DaemonEventLoopError eventLoopError ->+        unlines ["The daemon encountered an internal event loop error:", "", toS (displayException eventLoopError)]+      DaemonSocketError -> "There was an error with the socket"+      DaemonPushFailure -> "Failed to push some store paths"++-- | A wrapper for SomeException that implements equality.+newtype UnhandledException+  = UnhandledException SomeException+  deriving stock (Show)++instance Eq UnhandledException where+  UnhandledException e1 == UnhandledException e2 =+    displayException e1 == displayException e2++-- Convert to an error to an integer exit code++class HasExitCode a where+  toExitCode :: a -> ExitCode++  toExitCodeInt :: a -> Int+  toExitCodeInt x = case toExitCode x of+    ExitSuccess -> 0+    ExitFailure n -> n++instance HasExitCode DaemonError where+  toExitCode err = ExitFailure $ case err of+    DaemonPushFailure -> 3+    _remainingErrors -> 1++instance (HasExitCode a) => HasExitCode (Maybe a) where+  toExitCode Nothing = ExitSuccess+  toExitCode (Just e) = toExitCode e++instance (HasExitCode e) => HasExitCode (Either e a) where+  toExitCode (Right _) = ExitSuccess+  toExitCode (Left e) = toExitCode e
+ src/Cachix/Daemon/Types/EventLoop.hs view
@@ -0,0 +1,29 @@+module Cachix.Daemon.Types.EventLoop+  ( EventLoop (..),+    EventLoopError (..),+    ExitLatch,+  )+where++import Control.Concurrent.STM.TBMQueue (TBMQueue)+import Protolude++-- | An event loop that processes a queue of events.+data EventLoop event output = EventLoop+  { queue :: TBMQueue event,+    exitLatch :: ExitLatch output+  }++-- | An exit latch is a semaphore that signals the event loop to exit.+-- The exit code should be returned by the 'EventLoop'.+type ExitLatch a = MVar (Either EventLoopError a)++data EventLoopError+  = EventLoopClosed+  | EventLoopFull+  deriving stock (Show, Eq)++instance Exception EventLoopError where+  displayException = \case+    EventLoopClosed -> "The event loop is closed"+    EventLoopFull -> "The event loop is full"
+ src/Cachix/Daemon/Types/Log.hs view
@@ -0,0 +1,32 @@+module Cachix.Daemon.Types.Log+  ( LogLevel (..),+    Logger (..),+  )+where++import Katip qualified+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/Daemon/Types/PushEvent.hs view
@@ -0,0 +1,42 @@+module Cachix.Daemon.Types.PushEvent+  ( PushEvent (..),+    PushEventMessage (..),+    PushRetryStatus (..),+    newPushRetryStatus,+  )+where++import Cachix.Daemon.Protocol qualified as Protocol+import Control.Retry (RetryStatus (..))+import Data.Aeson (FromJSON, ToJSON)+import Data.Time (UTCTime)+import Protolude++data PushEvent = PushEvent+  { -- TODO: newtype a monotonic clock+    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/Daemon/Types/PushManager.hs view
@@ -0,0 +1,139 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}++module Cachix.Daemon.Types.PushManager+  ( PushManagerEnv (..),+    PushManager (..),+    PushJobStore,+    PushJob (..),+    JobStatus (..),+    JobStats (..),+    PushResult (..),+    OnPushEvent,+    Task (..),+    TimeoutOptions (..),+  )+where++import Cachix.Client.Push (PushParams)+import Cachix.Daemon.Log qualified as Log+import Cachix.Daemon.Protocol qualified as Protocol+import Cachix.Daemon.Types.Log (Logger)+import Cachix.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 Katip qualified+import Protolude++data Task+  = ResolveClosure Protocol.PushRequestId+  | PushStorePath FilePath++type PushJobStore = TVar (HashMap Protocol.PushRequestId PushJob)++type StorePathIndex = TVar (HashMap FilePath [Protocol.PushRequestId])++-- TODO: a lot of the logic surrounding deduping, search, and job tracking could be replaced by sqlite.+-- sqlite can run in-memory if we don't need persistence.+-- If we do, then we can we get stop/resume for free.+data PushManagerEnv = PushManagerEnv+  { pmPushParams :: PushParams PushManager (),+    -- | A store of push jobs indexed by a PushRequestId.+    pmPushJobs :: PushJobStore,+    -- | A mapping of store paths to push requests.+    -- Used when deduplicating pushes.+    pmStorePathIndex :: StorePathIndex,+    -- | FIFO queue of push tasks.+    pmTaskQueue :: TBMQueue Task,+    -- | A semaphore to control task concurrency.+    pmTaskSemaphore :: QSem,+    -- | A callback for push events.+    pmOnPushEvent :: OnPushEvent,+    -- | The timestamp of the most recent event. This is used to track activity internally.+    pmLastEventTimestamp :: TVar UTCTime,+    -- | The number of pending (uncompleted) jobs.+    pmPendingJobCount :: TVar Int,+    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,+      Alternative,+      MonadPlus+    )++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)++data TimeoutOptions = TimeoutOptions+  { -- | The maximum time to wait in seconds.+    toTimeout :: Float,+    -- | The interval at which to check the timeout condition.+    toPollingInterval :: Float+  }+  deriving stock (Eq, Show)
+ src/Cachix/Daemon/Types/SocketStore.hs view
@@ -0,0 +1,23 @@+module Cachix.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..)) where++import Control.Concurrent.STM.TVar (TVar)+import Data.HashMap.Strict (HashMap)+import Data.UUID (UUID)+import Network.Socket qualified as Network (Socket)+import Protolude++data Socket = Socket+  { socketId :: SocketId,+    socket :: Network.Socket,+    handlerThread :: Async ()+  }++instance Eq Socket where+  (==) = (==) `on` socketId++instance Ord Socket where+  compare = comparing socketId++type SocketId = UUID++newtype SocketStore = SocketStore (TVar (HashMap SocketId Socket))
+ src/Cachix/Daemon/Worker.hs view
@@ -0,0 +1,62 @@+module Cachix.Daemon.Worker+  ( startWorkers,+    stopWorkers,+    startWorker,+    stopWorker,+    Immortal.Thread,+  )+where++import Control.Concurrent.Async qualified as Async+import Control.Concurrent.STM.TBMQueue+import Control.Immortal qualified as Immortal+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Katip qualified+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/Deploy/Activate.hs view
@@ -3,27 +3,27 @@  module Cachix.Deploy.Activate where -import qualified Cachix.API.WebSocketSubprotocol as WSS-import qualified Cachix.Client.InstallationMode as InstallationMode-import qualified Cachix.Client.NetRc as NetRc+import Cachix.API.WebSocketSubprotocol qualified as WSS+import Cachix.Client.InstallationMode qualified as InstallationMode+import Cachix.Client.NetRc qualified as NetRc import Cachix.Client.URI (URI)-import qualified Cachix.Client.URI as URI-import qualified Cachix.Deploy.Log as Log-import qualified Cachix.Types.BinaryCache as BinaryCache+import Cachix.Client.URI qualified as URI+import Cachix.Deploy.Log qualified as Log+import Cachix.Types.BinaryCache qualified as BinaryCache import Cachix.Types.Permission (Permission (..))-import qualified Data.Aeson as Aeson+import Data.Aeson qualified as Aeson #if MIN_VERSION_aeson(2,0,0) import qualified Data.Aeson.KeyMap as HM #else import qualified Data.HashMap.Strict as HM #endif-import qualified Data.Conduit.Combinators as Conduit-import qualified Data.Conduit.Process as Conduit-import qualified Data.Vector as Vector+import Data.Conduit.Combinators qualified as Conduit+import Data.Conduit.Process qualified as Conduit+import Data.Vector qualified as Vector import Protolude hiding (log, toS) import Protolude.Conv (toS) import Servant.Auth.Client (Token (..))-import qualified System.Directory as Directory+import System.Directory qualified as Directory import System.FilePath ((</>)) import System.IO.Temp (withSystemTempDirectory) import System.Process
src/Cachix/Deploy/ActivateCommand.hs view
@@ -1,38 +1,38 @@ module Cachix.Deploy.ActivateCommand where -import qualified Cachix.API.Deploy.V1 as API.V1-import qualified Cachix.API.Deploy.V2 as API.V2+import Cachix.API.Deploy.V1 qualified as API.V1+import Cachix.API.Deploy.V2 qualified as API.V2 import Cachix.API.Error (escalate)-import qualified Cachix.API.WebSocketSubprotocol as WSS-import qualified Cachix.Client.Config as Config-import qualified Cachix.Client.Env as Env-import qualified Cachix.Client.Retry as Retry+import Cachix.API.WebSocketSubprotocol qualified as WSS+import Cachix.Client.Config qualified as Config+import Cachix.Client.Env qualified as Env+import Cachix.Client.Retry qualified as Retry import Cachix.Client.Servant (deployClientV1, deployClientV2)-import qualified Cachix.Client.URI as URI+import Cachix.Client.URI qualified as URI import Cachix.Client.Version (versionNumber)-import qualified Cachix.Deploy.OptionsParser as DeployOptions-import qualified Cachix.Deploy.Websocket as WebSocket+import Cachix.Deploy.OptionsParser qualified as DeployOptions+import Cachix.Deploy.Websocket qualified as WebSocket import Cachix.Types.Deploy (Deploy)-import qualified Cachix.Types.Deploy as Types-import qualified Cachix.Types.DeployResponse as DeployResponse-import qualified Cachix.Types.Deployment as Deployment-import qualified Control.Concurrent.Async as Async-import qualified Data.Aeson as Aeson+import Cachix.Types.Deploy qualified as Types+import Cachix.Types.DeployResponse qualified as DeployResponse+import Cachix.Types.Deployment qualified as Deployment+import Control.Concurrent.Async qualified as Async+import Data.Aeson qualified as Aeson import Data.HashMap.Strict (filterWithKey)-import qualified Data.HashMap.Strict as HM-import qualified Data.Text as Text-import qualified Data.Text.IO as Text+import Data.HashMap.Strict qualified as HM+import Data.Text qualified as Text+import Data.Text.IO qualified as Text import Data.UUID (UUID)-import qualified Data.UUID as UUID-import qualified Network.WebSockets as WS+import Data.UUID qualified as UUID+import Network.WebSockets qualified as WS import Protolude hiding (toS) import Protolude.Conv import Servant.Auth.Client (Token (..)) import Servant.Client.Streaming (ClientEnv, runClientM) import Servant.Conduit () import System.Environment (getEnv)-import qualified Text.Megaparsec as Parse-import qualified Text.Megaparsec.Char as Parse+import Text.Megaparsec qualified as Parse+import Text.Megaparsec.Char qualified as Parse  run :: Env.Env -> DeployOptions.ActivateOptions -> IO () run env DeployOptions.ActivateOptions {DeployOptions.payloadPath, DeployOptions.agents, DeployOptions.deployAsync} = do
src/Cachix/Deploy/Agent.hs view
@@ -3,39 +3,39 @@  module Cachix.Deploy.Agent where -import qualified Cachix.API.WebSocketSubprotocol as WSS-import qualified Cachix.Client.Config as Config+import Cachix.API.WebSocketSubprotocol qualified as WSS+import Cachix.Client.Config qualified as Config import Cachix.Client.URI (URI)-import qualified Cachix.Client.URI as URI+import Cachix.Client.URI qualified as URI import Cachix.Client.Version (versionNumber)-import qualified Cachix.Deploy.Deployment as Deployment-import qualified Cachix.Deploy.Lock as Lock-import qualified Cachix.Deploy.Log as Log-import qualified Cachix.Deploy.OptionsParser as CLI-import qualified Cachix.Deploy.StdinProcess as StdinProcess-import qualified Cachix.Deploy.Websocket as WebSocket-import qualified Control.Concurrent.Async as Async+import Cachix.Deploy.Deployment qualified as Deployment+import Cachix.Deploy.Lock qualified as Lock+import Cachix.Deploy.Log qualified as Log+import Cachix.Deploy.OptionsParser qualified as CLI+import Cachix.Deploy.StdinProcess qualified as StdinProcess+import Cachix.Deploy.Websocket qualified as WebSocket+import Control.Concurrent.Async qualified as Async import Control.Concurrent.Extra (once)-import qualified Control.Concurrent.MVar as MVar+import Control.Concurrent.MVar qualified as MVar import Control.Exception.Safe (onException)-import qualified Control.Exception.Safe as Safe-import qualified Control.Retry as Retry-import qualified Data.Aeson as Aeson+import Control.Exception.Safe qualified as Safe+import Control.Retry qualified as Retry+import Data.Aeson qualified as Aeson import Data.IORef import Data.String (String)-import qualified Katip as K+import Katip qualified as K import Paths_cachix (getBinDir) import Protolude hiding (onException, toS, (<.>)) import Protolude.Conv-import qualified System.Directory as Directory+import System.Directory qualified 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-import qualified System.Posix.Types as Posix-import qualified System.Posix.User as Posix.User-import qualified System.Timeout as Timeout+import System.Posix.Files qualified as Posix.Files+import System.Posix.Process qualified as Posix+import System.Posix.Signals qualified as Signals+import System.Posix.Types qualified as Posix+import System.Posix.User qualified as Posix.User+import System.Timeout qualified as Timeout  type ServiceWebSocket = WebSocket.WebSocket (WSS.Message WSS.AgentCommand) (WSS.Message WSS.BackendCommand)
src/Cachix/Deploy/Deployment.hs view
@@ -1,9 +1,9 @@ module Cachix.Deploy.Deployment where -import qualified Cachix.API.WebSocketSubprotocol as WSS+import Cachix.API.WebSocketSubprotocol qualified as WSS import Cachix.Client.URI as Cachix-import qualified Cachix.Deploy.Log as Log-import qualified Data.Aeson as Aeson+import Cachix.Deploy.Log qualified as Log+import Data.Aeson qualified as Aeson import Protolude  -- | Everything required for the standalone deployment binary to complete a
src/Cachix/Deploy/Lock.hs view
@@ -10,9 +10,9 @@   ) where -import qualified Lukko as Lock+import Lukko qualified as Lock import Protolude hiding ((<.>))-import qualified System.Directory as Directory+import System.Directory qualified as Directory import System.FilePath ((<.>), (</>)) import System.Posix (getProcessID) import System.Posix.Types (CPid (..))
src/Cachix/Deploy/Log.hs view
@@ -2,17 +2,17 @@  module Cachix.Deploy.Log where -import qualified Cachix.API.WebSocketSubprotocol as WSS-import qualified Control.Concurrent.STM.TMQueue as TMQueue+import Cachix.API.WebSocketSubprotocol qualified as WSS+import Control.Concurrent.STM.TMQueue qualified as TMQueue import Control.Exception.Safe (MonadMask, bracket)-import qualified Data.Aeson as Aeson+import Data.Aeson qualified as Aeson import Data.Conduit ((.|))-import qualified Data.Conduit as Conduit-import qualified Data.Conduit.Combinators as Conduit-import qualified Data.Conduit.TQueue as Conduit+import Data.Conduit qualified as Conduit+import Data.Conduit.Combinators qualified as Conduit+import Data.Conduit.TQueue qualified as Conduit import Data.Time.Clock (getCurrentTime)-import qualified Katip-import qualified Network.WebSockets as WS+import Katip qualified+import Network.WebSockets qualified as WS import Protolude hiding (bracket, toS) import Protolude.Conv (toS)
src/Cachix/Deploy/Websocket.hs view
@@ -3,38 +3,38 @@ -- | A high-level, multiple reader, single writer interface for Websocket clients. module Cachix.Deploy.Websocket where -import qualified Cachix.Client.Retry as Retry-import qualified Cachix.Client.URI as URI+import Cachix.Client.Retry qualified as Retry+import Cachix.Client.URI qualified as URI import Cachix.Client.Version (versionNumber)-import qualified Cachix.Deploy.Log as Log-import qualified Cachix.Deploy.WebsocketPong as WebsocketPong-import qualified Control.Concurrent.Async as Async-import qualified Control.Concurrent.MVar as MVar-import qualified Control.Concurrent.STM.TBMQueue as TBMQueue-import qualified Control.Concurrent.STM.TMChan as TMChan+import Cachix.Deploy.Log qualified as Log+import Cachix.Deploy.WebsocketPong qualified as WebsocketPong+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar qualified as MVar+import Control.Concurrent.STM.TBMQueue qualified as TBMQueue+import Control.Concurrent.STM.TMChan qualified as TMChan import Control.Exception.Safe (Handler (..), MonadMask, isSyncException)-import qualified Control.Exception.Safe as Safe-import qualified Control.Retry as Retry-import qualified Data.Aeson as Aeson-import qualified Data.ByteString as BS+import Control.Exception.Safe qualified as Safe+import Control.Retry qualified as Retry+import Data.Aeson qualified as Aeson+import Data.ByteString qualified as BS import Data.Conduit ((.|))-import qualified Data.Conduit as Conduit-import qualified Data.Conduit.Combinators as Conduit-import qualified Data.Conduit.TQueue as Conduit-import qualified Data.IORef as IORef+import Data.Conduit qualified as Conduit+import Data.Conduit.Combinators qualified as Conduit+import Data.Conduit.TQueue qualified as Conduit+import Data.IORef qualified as IORef import Data.String (String) import Data.String.Here (iTrim)-import qualified Data.Text as Text-import qualified Data.Time.Clock as Time-import qualified Katip as K-import qualified Network.HTTP.Simple as HTTP-import qualified Network.WebSockets as WS-import qualified Network.WebSockets.Connection as WS.Connection+import Data.Text qualified as Text+import Data.Time.Clock qualified as Time+import Katip qualified as K+import Network.HTTP.Simple qualified as HTTP+import Network.WebSockets qualified as WS+import Network.WebSockets.Connection qualified as WS.Connection import Protolude hiding (Handler, toS) import Protolude.Conv-import qualified System.Info-import qualified System.Timeout as Timeout-import qualified Wuss+import System.Info qualified+import System.Timeout qualified as Timeout+import Wuss qualified  -- | A reliable WebSocket connection that can be run ergonomically in a -- separate thread.
src/Cachix/Deploy/WebsocketPong.hs view
@@ -3,9 +3,9 @@ module Cachix.Deploy.WebsocketPong where  import Data.IORef (IORef)-import qualified Data.IORef as IORef+import Data.IORef qualified as IORef import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds)-import qualified Network.WebSockets as WS+import Network.WebSockets qualified as WS import Protolude  type LastPongState = IORef UTCTime
src/Data/Conduit/ByteString.hs view
@@ -1,7 +1,7 @@ module Data.Conduit.ByteString where  import Conduit (MonadUnliftIO)-import qualified Data.ByteString as BS+import Data.ByteString qualified as BS import Data.ByteString.Internal (ByteString (PS), mallocByteString) import Data.ByteString.Unsafe (unsafeIndex) import Data.Conduit (ConduitT, await, yield, (.|))
test/Daemon/PostBuildHookSpec.hs view
@@ -1,9 +1,9 @@ module Daemon.PostBuildHookSpec where -import Cachix.Client.Daemon.PostBuildHook+import Cachix.Daemon.PostBuildHook import Data.String import Protolude-import qualified System.Environment as System+import System.Environment qualified as System import Test.Hspec  spec :: Spec
+ test/Daemon/ProtocolSpec.hs view
@@ -0,0 +1,23 @@+module Daemon.ProtocolSpec where++import Cachix.Daemon.Protocol (splitMessages)+import Protolude+import Test.Hspec++spec :: Spec+spec = do+  describe "split messages" $ do+    forM_ testCases $ \(input, output) ->+      it (show input) $+        splitMessages input `shouldBe` output++testCases :: [(ByteString, ([ByteString], ByteString))]+testCases =+  [ ("hello\n", (["hello"], "")),+    ("hello\nthere\n", (["hello", "there"], "")),+    ("hello\nthere", (["hello"], "there")),+    ("", ([], "")),+    ("\n", ([], "")),+    ("\n\n", ([], "")),+    ("\nhello\n", (["hello"], ""))+  ]
test/Daemon/PushManagerSpec.hs view
@@ -1,19 +1,31 @@ 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.Env qualified as Env import Cachix.Client.OptionsParser (defaultPushOptions)+import Cachix.Client.Push (PushSecret (PushToken))+import Cachix.Daemon.Log qualified as Log+import Cachix.Daemon.Protocol qualified as Protocol+import Cachix.Daemon.Push qualified as Daemon.Push+import Cachix.Daemon.PushManager+import Cachix.Daemon.PushManager.PushJob qualified as PushJob+import Cachix.Daemon.Types.PushManager+import Cachix.Types.BinaryCache qualified as BinaryCache+import Cachix.Types.Permission (Permission (Write)) import Control.Concurrent.Async (concurrently_) import Control.Concurrent.STM.TVar+import Control.Monad (fail) import Control.Retry (defaultRetryStatus)-import qualified Data.Set as Set+import Data.Set qualified as Set import Data.Time (getCurrentTime)+import Hercules.CNix qualified as CNix import Protolude+import Servant.Auth.Client (Token (Token))+import System.IO.Temp (withSystemTempDirectory) import Test.Hspec +instance MonadFail PushManager where+  fail msg = liftIO (expectationFailure msg) >> mzero+ spec :: Spec spec = do   describe "push job" $ do@@ -79,19 +91,17 @@   describe "push manager" $ do     it "queues push jobs " $ inPushManager $ do       let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"]}-      pushId <- addPushJob request-      mpushJob <- lookupPushJob pushId+      Just pushId <- addPushJob request+      Just pushJob <- lookupPushJob pushId       liftIO $ do-        mpushJob `shouldSatisfy` isJust-        for_ mpushJob $ \pushJob -> do-          PushJob.pushId pushJob `shouldBe` pushId-          PushJob.pushRequest pushJob `shouldBe` request+        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+      Just pushId <- addPushJob pushRequest        let pathSet = Set.fromList paths           closure = PushJob.ResolvedClosure pathSet pathSet@@ -116,16 +126,13 @@             }      describe "graceful shutdown" $ do-      it "shuts down with no jobs" $ do-        logger <- Log.new "daemon" Nothing Log.Debug-        pm <- newPushManagerEnv defaultPushOptions logger mempty-        stopPushManager timeoutOptions pm+      it "shuts down with no jobs" $+        withPushManager $+          stopPushManager timeoutOptions -      it "shuts down after jobs complete" $ do+      it "shuts down after jobs complete" $ withPushManager $ \pm -> do         let paths = ["foo"]         let longTimeoutOptions = TimeoutOptions {toTimeout = 10.0, toPollingInterval = 0.1}-        logger <- Log.new "daemon" Nothing Log.Debug-        pm <- newPushManagerEnv defaultPushOptions logger mempty          _ <- runPushManager pm $ do           let request = Protocol.PushRequest {Protocol.storePaths = paths}@@ -135,14 +142,13 @@           runPushManager pm $             for_ paths pushStorePathDone -      it "shuts down on job stall" $ do-        logger <- Log.new "daemon" Nothing Log.Debug-        pm <- newPushManagerEnv defaultPushOptions logger mempty-        _ <- runPushManager pm $ do-          let request = Protocol.PushRequest {Protocol.storePaths = ["foo"]}-          addPushJob request+      it "shuts down on job stall" $+        withPushManager $ \pm -> do+          _ <- runPushManager pm $ do+            let request = Protocol.PushRequest {Protocol.storePaths = ["foo"]}+            addPushJob request -        stopPushManager timeoutOptions pm+          stopPushManager timeoutOptions pm    describe "STM" $     describe "timeout" $ do@@ -152,11 +158,36 @@  withPushManager :: (PushManagerEnv -> IO a) -> IO a withPushManager f = do-  logger <- liftIO $ Log.new "daemon" Nothing Log.Debug-  newPushManagerEnv defaultPushOptions logger mempty >>= f+  CNix.init+  withTempStore $ \store -> do+    logger <- liftIO $ Log.new "daemon" Nothing Log.Debug+    cachixOptions <- Env.defaultCachixOptions+    clientEnv <- Env.createClientEnv cachixOptions+    let binaryCache = newBinaryCache "test"+        pushSecret = PushToken (Token "test")+        pushOptions = defaultPushOptions+        pushParams = Daemon.Push.newPushParams store clientEnv binaryCache pushSecret pushOptions+    newPushManagerEnv pushOptions pushParams mempty logger >>= f  inPushManager :: PushManager a -> IO a inPushManager f = withPushManager (`runPushManager` f)++withTempStore :: (CNix.Store -> IO a) -> IO a+withTempStore f =+  withSystemTempDirectory "cnix-test-store" $ \d ->+    CNix.withStoreFromURI (toS d) f++newBinaryCache :: BinaryCache.BinaryCacheName -> BinaryCache.BinaryCache+newBinaryCache name =+  BinaryCache.BinaryCache+    { BinaryCache.name = name,+      BinaryCache.uri = "https://" <> name <> ".cachix.org",+      BinaryCache.isPublic = True,+      BinaryCache.publicSigningKeys = [],+      BinaryCache.githubUsername = "",+      BinaryCache.permission = Write,+      BinaryCache.preferredCompressionMethod = BinaryCache.ZSTD+    }  timeoutOptions :: TimeoutOptions timeoutOptions = TimeoutOptions {toTimeout = 0.2, toPollingInterval = 0.1}
test/DeploySpec.hs view
@@ -1,11 +1,11 @@ module DeploySpec where -import qualified Cachix.Client.Config as Config+import Cachix.Client.Config qualified 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 Cachix.Deploy.Log qualified as Log+import Cachix.Deploy.OptionsParser qualified as CLI+import Control.Retry qualified as Retry import Protolude import System.IO.Temp (withSystemTempDirectory) import Test.Hspec
test/InstallationModeSpec.hs view
@@ -1,7 +1,7 @@ module InstallationModeSpec where  import Cachix.Client.InstallationMode-import qualified Cachix.Client.NixConf as NixConf+import Cachix.Client.NixConf qualified as NixConf import Protolude import Test.Hspec
test/Main.hs view
@@ -1,7 +1,7 @@ module Main where  import Protolude-import qualified Spec+import Spec qualified import Test.Hspec.Runner  main :: IO ()
test/NetRcSpec.hs view
@@ -1,6 +1,6 @@ module NetRcSpec where -import qualified Cachix.Client.NetRc as NetRc+import Cachix.Client.NetRc qualified as NetRc import Cachix.Types.BinaryCache (BinaryCache (..), CompressionMethod (..)) import Cachix.Types.Permission (Permission (..)) import Protolude
test/URISpec.hs view
@@ -1,13 +1,13 @@ module URISpec where -import qualified Cachix.Client.URI as URI-import qualified Data.Aeson as Aeson-import qualified Data.ByteString.Lazy as BL+import Cachix.Client.URI qualified as URI+import Data.Aeson qualified as Aeson+import Data.ByteString.Lazy qualified as BL import Data.Either.Extra-import qualified Dhall-import qualified Dhall.Core+import Dhall qualified+import Dhall.Core qualified import Protolude-import qualified Servant.Client.Core as Servant+import Servant.Client.Core qualified as Servant import Test.Hspec  secureScheme :: ByteString