packages feed

cachix 1.11.0 → 1.11.1

raw patch · 18 files changed

+447/−235 lines, 18 filesdep +http-media

Dependencies added: http-media

Files

CHANGELOG.md view
@@ -4,6 +4,22 @@  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.11.1] - 2026-04-29++### Added++- client: support reading the retry_after field in "problem details" response bodies++### Changed++- client: increase the default number of retries for HTTP requests+- client: always check for the Retry-After header in error responses, not just for certain status codes++### Fixed++- client: add missing retries to backend endpoints to help handle intermittent network issues+- daemon: remove risk of race conditions when checking for job completion+- daemon: fail any pending jobs on shutdown timeout  ## [1.11.0] - 2026-03-20
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.11.0+version:            1.11.1 synopsis:   Command-line client for Nix binary cache hosting https://cachix.org @@ -161,6 +161,7 @@     , http-client     , http-client-tls     , http-conduit+    , http-media     , http-types     , immortal     , inline-c-cpp
src/Cachix/Client/Command/Cache.hs view
@@ -8,18 +8,17 @@ 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.Retry (retryClientM) 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+  res <- retryClientM (clientenv env) $ API.getCache cachixClient token name   case res of     Left err -> Push.handleCacheResponse name optionalAuthToken err     Right binaryCache -> do
src/Cachix/Client/Command/Config.hs view
@@ -7,7 +7,7 @@ import Cachix.API.Error import Cachix.Client.Config qualified as Config import Cachix.Client.Env (Env (..))-import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Retry (retryClientM) import Cachix.Client.Secrets   ( SigningKey (SigningKey),     exportSigningKey,@@ -23,7 +23,6 @@ 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 ()@@ -42,9 +41,8 @@       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+    escalate+      =<< retryClientM (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]
src/Cachix/Client/Command/Doctor.hs view
@@ -4,7 +4,7 @@ import Cachix.Client.Config qualified as Config import Cachix.Client.Env (Env (..)) import Cachix.Client.OptionsParser (DaemonOptions (..), DoctorOptions (..))-import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Retry (retryClientM) import Cachix.Client.Servant (cachixClient, isErr) import Cachix.Daemon.Listen (getSocketPath) import Cachix.Daemon.Protocol qualified as Protocol@@ -21,7 +21,7 @@ import Protolude hiding (toS) import Protolude.Conv import Servant.Auth.Client-import Servant.Client.Streaming (ClientError, runClientM)+import Servant.Client.Streaming (ClientError) import System.Directory (doesFileExist) import System.Environment (lookupEnv) import System.Timeout (timeout)@@ -220,7 +220,7 @@   let hasSigningKey = any (\c -> Config.name c == cacheName && not (T.null (Config.secretKey c))) configuredCaches    -- Try to get cache info (validates auth and connectivity)-  cacheRes <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token cacheName+  cacheRes <- retryClientM (clientenv env) $ API.getCache cachixClient token cacheName    case cacheRes of     Right bc ->@@ -343,7 +343,7 @@           }     Just storeHash -> do       -- Use narinfoBulk to check if the path exists-      result <- retryHttp $ (`runClientM` clientenv env) $ API.narinfoBulk cachixClient token cacheName [storeHash]+      result <- retryClientM (clientenv env) $ API.narinfoBulk cachixClient token cacheName [storeHash]       case result of         Right missingHashes ->           let status = if storeHash `elem` missingHashes then NotInCache else InCache
src/Cachix/Client/Command/Pin.hs view
@@ -7,14 +7,13 @@ import Cachix.Client.Env (Env (..)) import Cachix.Client.Exception (CachixException (..)) import Cachix.Client.OptionsParser (PinOptions (..))-import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Retry (retryClientM) 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 ()@@ -36,9 +35,8 @@             keep = pinKeep pinOpts           }   void $-    escalate <=< retryHttp $-      (`runClientM` clientenv env) $-        API.createPin cachixClient authToken (pinCacheName pinOpts) pinCreate+    escalate+      =<< retryClientM (clientenv env) (API.createPin cachixClient authToken (pinCacheName pinOpts) pinCreate)   where     validateArtifact :: Text -> Text -> IO ()     validateArtifact storePath artifact = do
src/Cachix/Client/Command/Push.hs view
@@ -19,7 +19,7 @@ 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.Retry (retryClientM) import Cachix.Client.Secrets import Cachix.Client.Servant import Cachix.Types.BinaryCache (BinaryCacheName)@@ -137,7 +137,7 @@   compressionMethodBackend <- case pushSecret of     PushSigningKey {} -> pure Nothing     PushToken token -> do-      res <- retryHttp $ (`runClientM` clientenv env) $ API.getCache cachixClient token name+      res <- retryClientM (clientenv env) $ API.getCache cachixClient token name       case res of         Left err -> handleCacheResponse name authToken err         Right binaryCache -> pure (Just $ BinaryCache.preferredCompressionMethod binaryCache)
src/Cachix/Client/Command/Watch.hs view
@@ -7,6 +7,7 @@ where  import Cachix.Client.Command.Push+import Cachix.Client.Config qualified as Config import Cachix.Client.Env (Env (..)) import Cachix.Client.InstallationMode qualified as InstallationMode import Cachix.Client.OptionsParser@@ -19,6 +20,7 @@ import Cachix.Client.Push import Cachix.Client.WatchStore qualified as WatchStore import Cachix.Daemon qualified as Daemon+import Cachix.Daemon.Log qualified as Daemon.Log import Cachix.Daemon.NarinfoQuery (NarinfoQueryOptions) import Cachix.Daemon.PostBuildHook qualified as Daemon.PostBuildHook import Cachix.Daemon.Progress qualified as Daemon.Progress@@ -87,9 +89,14 @@                   daemonKeepAliveInterval = defaultKeepAliveInterval,                   daemonKeepAliveTimeout = defaultKeepAliveTimeout                 }-        daemon <- Daemon.new env store daemonOptions (Just logHandle) pushOpts cacheName+        let logLevel =+              if Config.verbose (cachixoptions env)+                then Daemon.Log.Debug+                else Daemon.Log.Info+        logger <- Daemon.Log.new Daemon.Log.namespace (Just logHandle) logLevel+        exitCode <- Daemon.Log.withLogger logger $ \registeredLogger -> do+          daemon <- Daemon.new env store daemonOptions registeredLogger pushOpts cacheName -        exitCode <-           bracket (startDaemonThread daemon) (shutdownDaemonThread daemon logHandle) $ \_ -> do             processEnv <- getEnvironment             let newProcessEnv = Daemon.PostBuildHook.modifyEnv (Daemon.PostBuildHook.envVar hookEnv) processEnv
src/Cachix/Client/Push.hs view
@@ -49,7 +49,7 @@ import Cachix.Client.CNix (formatStorePathWarning, validateStorePath) import Cachix.Client.Exception (CachixException (..)) import Cachix.Client.Push.S3 qualified as Push.S3-import Cachix.Client.Retry (retryAll, retryHttp)+import Cachix.Client.Retry (retryAll, retryClientM, retryHttp) import Cachix.Client.Secrets import Cachix.Client.Servant import Cachix.Types.BinaryCache qualified as BinaryCache@@ -188,13 +188,12 @@ narinfoExists pushParams storeHash = do   let cacheName = pushParamsName pushParams       authToken = getCacheAuthToken (pushParamsSecret pushParams)-  retryHttp $-    (`runClientM` pushParamsClientEnv pushParams) $-      API.narinfoHead-        cachixClient-        authToken-        cacheName-        (NarInfoHash.NarInfoHash (decodeUtf8With lenientDecode storeHash))+  retryClientM (pushParamsClientEnv pushParams) $+    API.narinfoHead+      cachixClient+      authToken+      cacheName+      (NarInfoHash.NarInfoHash (decodeUtf8With lenientDecode storeHash))  getCacheAuthToken :: PushSecret -> Token getCacheAuthToken (PushToken token) = token
src/Cachix/Client/Push/S3.hs view
@@ -12,7 +12,7 @@  import Cachix.API qualified as API import Cachix.API.Error-import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Retry (retryClientM, retryHttp) import Cachix.Client.Servant (cachixClient) import Cachix.Types.BinaryCache import Cachix.Types.MultipartUpload qualified as Multipart@@ -93,7 +93,7 @@     createMultipartUpload :: ConduitT ByteString Void m (Either ClientError Multipart.CreateMultipartUploadResponse)     createMultipartUpload = do       let createNarRequest = API.createNar cachixClient authToken cacheName (Just (compressionMethod options))-      liftIO $ retryHttp $ runClientM createNarRequest env+      liftIO $ retryClientM env createNarRequest      uploadPart :: UUID -> Text -> (Int, ByteString) -> m (Maybe Multipart.CompletedPart)     uploadPart narId uploadId (partNumber, !part) = do
src/Cachix/Client/Retry.hs view
@@ -1,19 +1,25 @@ module Cachix.Client.Retry-  ( retryAll,-    retryAllWithPolicy,+  ( -- * Whole-operation retry+    retryAll,++    -- * HTTP retry     retryHttp,-    retryHttpWith,-    endlessRetryPolicy,+    retryClientM,++    -- * Policies     endlessConstantRetryPolicy,   ) where -import Cachix.Client.Exception (CachixException (..))+import Cachix.Client.Exception (CachixException) 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.Aeson qualified as Aeson+import Data.ByteString.Lazy qualified as LBS import Data.List (lookup)+import Data.Text qualified as T import Data.Time   ( UTCTime,     defaultTimeLocale,@@ -25,31 +31,48 @@   ) import GHC.Read (Read (readPrec)) import Network.HTTP.Client qualified as HTTP-import Network.HTTP.Types.Header (hRetryAfter)+import Network.HTTP.Media qualified as Media+import Network.HTTP.Types.Header (Header, hContentType, hRetryAfter) import Network.HTTP.Types.Status qualified as HTTP import Protolude hiding (Handler (..), handleJust)-import Servant.Client (ClientError (..))+import Servant.Client (ClientEnv, ClientError (..)) import Servant.Client qualified as Servant+import Servant.Client.Streaming (ClientM, runClientM) import Text.ParserCombinators.ReadPrec qualified as ReadPrec (lift) -defaultRetryPolicy :: RetryPolicy-defaultRetryPolicy =-  exponentialBackoff (1000 * 1000) <> limitRetries 5+-- Policies +-- | Per-HTTP-request policy. Wall-time bounded so the user-visible delay is+-- predictable regardless of how many attempts fit. Jittered exponential+-- backoff to avoid thundering-herd on shared origins (CDN incidents).+httpRetryPolicy :: (MonadIO m) => RetryPolicyM m+httpRetryPolicy =+  limitRetriesByCumulativeDelay (secondsToMicros 90) $+    capDelay (secondsToMicros 15) $+      fullJitterBackoff (secondsToMicros 1)++-- | Whole-operation policy used by 'retryAll'. Small attempt count with a+-- short delay, since each attempt re-runs the entire operation (e.g. a full+-- NAR re-upload). Compounds with 'httpRetryPolicy' on HTTP errors, so+-- keep this tight to bound worst-case wall time per store path.+retryAllPolicy :: RetryPolicy+retryAllPolicy =+  constantDelay (secondsToMicros 5) <> limitRetries 3++-- | Constant 1s delay, no attempt cap. For callers that own their own+-- termination condition (e.g. websocket reconnect loops). endlessConstantRetryPolicy :: RetryPolicy endlessConstantRetryPolicy =-  constantDelay (1000 * 1000)+  constantDelay (secondsToMicros 1) -endlessRetryPolicy :: RetryPolicy-endlessRetryPolicy =-  exponentialBackoff (1000 * 1000)+-- Whole-operation retry +-- | Retry an entire operation on any synchronous exception, except for+-- 'ExitCode' and 'CachixException' (which are fatal). Use for operations+-- that re-run the full unit of work, like a NAR upload. retryAll :: (MonadIO m, MonadMask m) => (RetryStatus -> m a) -> m a-retryAll = retryAllWithPolicy defaultRetryPolicy--retryAllWithPolicy :: (MonadIO m, MonadMask m) => RetryPolicyM m -> (RetryStatus -> m a) -> m a-retryAllWithPolicy policy f =-  recovering policy handlers $+retryAll f =+  recovering retryAllPolicy handlers $     rethrowLinkedThreadExceptions . f   where     handlers = skipAsyncExceptions ++ [exitCodeHandler, cachixExceptionsHandler, allHandler]@@ -63,83 +86,43 @@     -- Retry everything else     allHandler _ = Handler $ \(_ :: SomeException) -> return True --- | Unwrap 'Async.ExceptionInLinkedThread' exceptions and rethrow the inner exception.-rethrowLinkedThreadExceptions :: (MonadCatch m) => m a -> m a-rethrowLinkedThreadExceptions =-  handleJust unwrapLinkedThreadException throwM--unwrapLinkedThreadException :: SomeException -> Maybe SomeException-unwrapLinkedThreadException e-  | Just (Async.ExceptionInLinkedThread _ e') <- fromException e = Just e'-  | otherwise = Nothing--retryHttp :: (MonadIO m, MonadMask m) => m a -> m a-retryHttp = retryHttpWith defaultRetryPolicy+-- HTTP retry --- | Retry policy for HTTP requests.+-- | Retry an HTTP operation that signals failure by throwing+-- 'HTTP.HttpException' or Servant 'ClientError'. Honors @Retry-After@ headers+-- and Cloudflare problem-details @retry_after@ bodies. ----- Retries a subset of HTTP exceptions and overrides the delay with the Retry-After header if present.-retryHttpWith :: forall m a. (MonadIO m, MonadMask m) => RetryPolicyM m -> m a -> m a-retryHttpWith policy = recoveringDynamic policy handlers . const+-- For Servant calls that return @Either ClientError a@ (e.g. 'runClientM'),+-- use 'retryClientM' instead — 'retryHttp' only triggers on thrown errors.+retryHttp :: (MonadIO m, MonadMask m) => m a -> m a+retryHttp = recoveringDynamic httpRetryPolicy handlers . const   where-    handlers :: [RetryStatus -> Handler m RetryAction]     handlers =-      skipAsyncExceptions' ++ [retryHttpExceptions, retryClientExceptions]--    skipAsyncExceptions' = map (fmap toRetryAction .) skipAsyncExceptions--    retryHttpExceptions _ = Handler httpExceptionToRetryHandler-    retryClientExceptions _ = Handler clientExceptionToRetryHandler--    httpExceptionToRetryHandler :: HTTP.HttpException -> m RetryAction-    httpExceptionToRetryHandler (HTTP.HttpExceptionRequest _ (HTTP.StatusCodeException response _))-      | statusMayHaveRetryHeader (HTTP.responseStatus response) = overrideDelayWithRetryAfter response-    httpExceptionToRetryHandler ex = return . toRetryAction . shouldRetryHttpException $ ex--    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-  | RetryAfterSeconds Int-  deriving (Eq, Show)--instance Read RetryAfter where-  readPrec = parseSeconds <|> parseWebDate-    where-      parseSeconds = RetryAfterSeconds <$> readPrec-      parseWebDate = ReadPrec.lift $ RetryAfterDate <$> readPTime True defaultTimeLocale rfc822DateFormat+      map (fmap toRetryAction .) skipAsyncExceptions+        ++ [ \_ -> Handler httpExceptionToRetryAction,+             \_ -> Handler clientErrorToRetryAction+           ] -overrideDelayWithRetryAfter :: (MonadIO m) => HTTP.Response a -> m RetryAction-overrideDelayWithRetryAfter response =-  case lookupRetryAfter response of-    Nothing ->-      return ConsultPolicy-    Just (RetryAfterSeconds seconds) ->-      return $ ConsultPolicyOverrideDelay (seconds * 1000 * 1000)-    Just (RetryAfterDate date) -> do-      seconds <- secondsFromNow date-      return $-        if seconds > 0-          then ConsultPolicyOverrideDelay (seconds * 1000 * 1000)-          else ConsultPolicy+-- | Run a Servant client action with HTTP retries.+--+-- 'runClientM' returns 'Either ClientError a' instead of throwing, so wrapping+-- it with 'retryHttp' would never trigger the retry handlers. This inspects+-- the return value directly, sharing the same retry decision logic as+-- 'retryHttp'.+retryClientM :: (MonadIO m, NFData a) => ClientEnv -> ClientM a -> m (Either ClientError a)+retryClientM env req =+  retryingDynamic httpRetryPolicy checkResult $ const $ liftIO (runClientM req env)   where-    secondsFromNow date = do-      now <- liftIO getCurrentTime-      return $ ceiling $ nominalDiffTimeToSeconds (date `diffUTCTime` now)--    lookupRetryAfter = readMaybe . decodeUtf8 <=< lookup hRetryAfter . HTTP.responseHeaders+    checkResult _ (Right _) = pure DontRetry+    checkResult _ (Left err) = clientErrorToRetryAction err --- | Determine whether the HTTP exception is worth retrying.+-- Retry decisions ----- Temporary connection or network transfer errors are good candidates.+-- Decide whether a given failure is worth retrying, and at what delay. Shared+-- between 'retryHttp' (exception path) and 'retryClientM' (Either path).++-- | Whether an HTTP exception is worth retrying. Temporary connection or+-- network transfer errors are good candidates. shouldRetryHttpException :: HTTP.HttpException -> Bool shouldRetryHttpException (HTTP.InvalidUrlException _ _) = False shouldRetryHttpException (HTTP.HttpExceptionRequest _ reason) =@@ -162,11 +145,124 @@     HTTP.HttpZlibException _ -> True     _ -> False --- | Determine whether the HTTP status code is worth retrying.+-- | Whether an HTTP status code is worth retrying. 429 and any 5xx. shouldRetryHttpStatusCode :: HTTP.Status -> Bool-shouldRetryHttpStatusCode code | code == HTTP.tooManyRequests429 = True-shouldRetryHttpStatusCode code | HTTP.statusIsServerError code = True-shouldRetryHttpStatusCode _ = False+shouldRetryHttpStatusCode code+  | code == HTTP.tooManyRequests429 = True+  | HTTP.statusIsServerError code = True+  | otherwise = False -statusMayHaveRetryHeader :: HTTP.Status -> Bool-statusMayHaveRetryHeader = flip elem [HTTP.tooManyRequests429, HTTP.serviceUnavailable503]+-- | Convert an 'HTTP.HttpException' to a 'RetryAction', honoring any 'Retry-After' hints.+httpExceptionToRetryAction :: (MonadIO m) => HTTP.HttpException -> m RetryAction+httpExceptionToRetryAction ex@(HTTP.HttpExceptionRequest _ (HTTP.StatusCodeException response body))+  | shouldRetryHttpException ex =+      let headers = HTTP.responseHeaders response+       in overrideDelayWithRetryAfter $+            lookupRetryAfter headers+              <|> lookupRetryAfterBody headers (LBS.fromStrict body)+httpExceptionToRetryAction ex = pure $ toRetryAction $ shouldRetryHttpException ex++-- | Convert a 'ClientError' to a 'RetryAction', honoring any 'Retry-After' hints.+clientErrorToRetryAction :: (MonadIO m) => ClientError -> m RetryAction+clientErrorToRetryAction (FailureResponse _req res)+  | shouldRetryHttpStatusCode (Servant.responseStatusCode res) =+      let headers = toList (Servant.responseHeaders res)+       in overrideDelayWithRetryAfter $+            lookupRetryAfter headers+              <|> lookupRetryAfterBody headers (Servant.responseBody res)+clientErrorToRetryAction (ConnectionError ex) =+  case fromException ex of+    Just httpException -> httpExceptionToRetryAction httpException+    Nothing -> pure DontRetry+clientErrorToRetryAction _ = pure DontRetry++-- Retry-After+--+-- Parse and honor server hints about when to retry,+data RetryAfter+  = RetryAfterDate UTCTime+  | RetryAfterSeconds Int+  deriving (Eq, Show)++instance Read RetryAfter where+  readPrec = parseSeconds <|> parseWebDate+    where+      parseSeconds = RetryAfterSeconds <$> readPrec+      parseWebDate = ReadPrec.lift $ RetryAfterDate <$> readPTime True defaultTimeLocale rfc822DateFormat++-- | Honor a parsed 'RetryAfter', clamped to 'maxRetryAfterMicros'. Negative+-- or past-dated values fall back to the policy's own delay.+overrideDelayWithRetryAfter :: (MonadIO m) => Maybe RetryAfter -> m RetryAction+overrideDelayWithRetryAfter Nothing = pure ConsultPolicy+overrideDelayWithRetryAfter (Just (RetryAfterSeconds s))+  | s <= 0 = pure ConsultPolicy+  | otherwise = pure $ ConsultPolicyOverrideDelay (clampMicros (secondsToMicros s))+overrideDelayWithRetryAfter (Just (RetryAfterDate date)) = do+  s <- secondsFromNow date+  pure $+    if s > 0+      then ConsultPolicyOverrideDelay (clampMicros (secondsToMicros s))+      else ConsultPolicy+  where+    secondsFromNow d = do+      now <- liftIO getCurrentTime+      pure $ ceiling $ nominalDiffTimeToSeconds (d `diffUTCTime` now)++-- | Bound a delay (in microseconds) by 'maxRetryAfterMicros'.+clampMicros :: Int -> Int+clampMicros micros+  | micros < 0 = maxRetryAfterMicros+  | otherwise = min micros maxRetryAfterMicros++lookupRetryAfter :: [Header] -> Maybe RetryAfter+lookupRetryAfter = readMaybe . T.unpack . decodeUtf8With lenientDecode <=< lookup hRetryAfter++-- | Parse @retry_after@ (in seconds) from an @application/problem+json@ body.+--+-- @retry_after@ is not part of RFC 9457 (which only standardizes @type@,+-- @status@, @title@, @detail@, @instance@), but Cloudflare returns it as an+-- extension member on origin 5xx error pages instead of the canonical+-- 'Retry-After' HTTP header. Honoring it lets us back off the right amount+-- during transient origin outages.+lookupRetryAfterBody :: [Header] -> LBS.ByteString -> Maybe RetryAfter+lookupRetryAfterBody headers body = do+  ct <- lookup hContentType headers+  -- Some origins serve problem-details bodies with the bare @application/json@+  -- content type, so accept that too.+  _ <- Media.matchContent acceptableContentTypes ct+  pd <- Aeson.decode' body+  RetryAfterSeconds <$> problemRetryAfter pd+  where+    acceptableContentTypes =+      ["application" Media.// "problem+json", "application" Media.// "json"]++newtype ProblemDetails = ProblemDetails {problemRetryAfter :: Maybe Int}++instance Aeson.FromJSON ProblemDetails where+  parseJSON = Aeson.withObject "ProblemDetails" $ \o ->+    ProblemDetails <$> o Aeson..:? "retry_after"++-- Time++-- | Convert seconds to microseconds (the unit 'threadDelay' and the @retry@+-- package use for delays).+secondsToMicros :: Int -> Int+secondsToMicros = (* 1_000_000)++-- | Cap on a server-supplied 'Retry-After' override. Bounds the per-attempt+-- sleep so a hostile or buggy origin can't park the client for a long time.+maxRetryAfterMicros :: Int+maxRetryAfterMicros = secondsToMicros 120++-- Linked-thread exception unwrapping++-- | Unwrap 'Async.ExceptionInLinkedThread' and rethrow the inner exception so+-- retry handlers can match on the actual cause.+rethrowLinkedThreadExceptions :: (MonadCatch m) => m a -> m a+rethrowLinkedThreadExceptions =+  handleJust unwrapLinkedThreadException throwM++unwrapLinkedThreadException :: SomeException -> Maybe SomeException+unwrapLinkedThreadException e+  | Just (Async.ExceptionInLinkedThread _ e') <- fromException e = Just e'+  | otherwise = Nothing
src/Cachix/Daemon.hs view
@@ -19,7 +19,7 @@ import Cachix.Client.OptionsParser (DaemonOptions, PushOptions, daemonNarinfoQueryOptions) import Cachix.Client.OptionsParser qualified as Options import Cachix.Client.Push-import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Retry (retryClientM) import Cachix.Client.Servant (cachixClient) import Cachix.Daemon.EventLoop qualified as EventLoop import Cachix.Daemon.Listen as Listen@@ -36,7 +36,7 @@ import Cachix.Types.BinaryCache (BinaryCacheName) import Cachix.Types.BinaryCache qualified as BinaryCache import Control.Concurrent.STM.TMChan-import Control.Exception.Safe (catchAny)+import Control.Exception.Safe (catchAny, tryAny) import Data.IORef (IORef, atomicModifyIORef', newIORef) import Data.Text qualified as T import Hercules.CNix.Store (Store, withStore)@@ -45,16 +45,19 @@ import Network.Socket qualified as Socket import Network.Socket.ByteString qualified as Socket.BS import Protolude hiding (bracket)-import Servant.Client.Streaming (runClientM) import System.Environment (lookupEnv) import System.IO.Error (isResourceVanishedError) import System.Posix.Process (getProcessID) import System.Posix.Signals qualified as Signal+import System.Timeout (timeout) import UnliftIO (MonadUnliftIO, withRunInIO) import UnliftIO.Async qualified as Async import UnliftIO.Exception (bracket)  -- | Configure a new daemon. Use 'run' to start it.+--+-- The caller must provide a logger with scribes already registered (e.g. by+-- wrapping the call in 'Log.withLogger'). new ::   -- | The Cachix environment.   Env ->@@ -62,20 +65,16 @@   Store ->   -- | Daemon-specific options.   DaemonOptions ->-  -- | An optional handle to output logs to.-  Maybe Handle ->+  -- | A logger with registered scribes.+  Log.Logger ->   -- | 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+new daemonEnv nixStore daemonOptions daemonLogger daemonPushOptions daemonCacheName = do+  Log.logMsg daemonLogger Katip.InfoS "Starting Cachix Daemon"    daemonEventLoop <- EventLoop.new   daemonPid <- getProcessID@@ -89,7 +88,7 @@    daemonPushSecret <- Command.Push.getPushSecretRequired (config daemonEnv) daemonCacheName   let authToken = getAuthTokenFromPushSecret daemonPushSecret-  daemonBinaryCache <- Push.getBinaryCache daemonEnv authToken daemonCacheName+  daemonBinaryCache <- Push.getBinaryCache daemonLogger daemonEnv authToken daemonCacheName    daemonSubscriptionManagerThread <- newEmptyMVar   daemonSubscriptionManager <- Subscription.newSubscriptionManager@@ -111,15 +110,20 @@ start :: Env -> DaemonOptions -> PushOptions -> BinaryCacheName -> IO () start daemonEnv daemonOptions daemonPushOptions daemonCacheName =   withStore $ \store -> do-    daemon <- new daemonEnv store daemonOptions Nothing daemonPushOptions daemonCacheName-    void $ runDaemon daemon installSignalHandlers-    result <- run daemon+    let logLevel =+          if Config.verbose (Env.cachixoptions daemonEnv)+            then Debug+            else Info+    logger <- Log.new Log.namespace Nothing logLevel+    result <- Log.withLogger logger $ \registeredLogger -> do+      daemon <- new daemonEnv store daemonOptions registeredLogger daemonPushOptions daemonCacheName+      void $ runDaemon daemon installSignalHandlers+      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@@ -185,7 +189,7 @@           authResult <- case maybeToken of             Nothing -> pure $ Left "No auth token configured"             Just token -> do-              res <- liftIO $ retryHttp $ (`runClientM` clientenv daemonEnv) $ API.getCache cachixClient token daemonCacheName+              res <- liftIO $ retryClientM (clientenv daemonEnv) $ API.getCache cachixClient token daemonCacheName               pure $ case res of                 Right _ -> Right ()                 Left err -> Left (show err)@@ -330,20 +334,18 @@   DaemonEnv {..} <- ask    -- 1. Drain: set latch (reject new jobs), wait for in-flight jobs to complete-  drainWithLogging daemonPushManager--  -- 2. Stop the batch processor (safe: all jobs completed the full pipeline)-  liftIO $ PushManager.stopBatchProcessor daemonPushManager--  -- 3. Close the task queue-  liftIO $ PushManager.closePushManager daemonPushManager--  -- 4. Stop worker threads (workers see closed queue and exit)-  withTakeMVar daemonWorkerThreads Worker.stopWorkers+  drained <- drainWithLogging daemonPushManager -  -- 5. Close all event subscriptions-  withTakeMVar daemonSubscriptionManagerThread (shutdownSubscriptions daemonSubscriptionManager)+  -- Fail any remaining jobs that didn't finish after the drain timeout+  unless drained $ do+    stuck <-+      PushManager.runPushManager daemonPushManager $+        PushManager.failPendingJobs "Daemon stopped before this path finished uploading"+    Katip.logFM Katip.WarningS $+      Katip.logStr+        ("Failed " <> show (length stuck) <> " stuck jobs after drain timeout." :: Text) +  -- 2. Compute the exit result now so clients see the real exit code, even if worker cleanup below blocks on a stuck upload.   failedJobs <-     PushManager.runPushManager daemonPushManager PushManager.getFailedPushJobs   let pushResult =@@ -351,9 +353,22 @@           then Right ()           else Left DaemonPushFailure -  -- 6. Gracefully close open connections to clients+  -- 3. Stop the batch processor and close the task queue so workers exit after their current task.+  liftIO $ PushManager.stopBatchProcessor daemonPushManager+  liftIO $ PushManager.closePushManager daemonPushManager++  -- 4. Disconnect clients with a goodbye message that includes the push result   Async.mapConcurrently_ (sayGoodbye daemonClients pushResult) =<< SocketStore.toList daemonClients +  -- 5. Stop worker threads. If drain succeeded, workers are idle and exit+  --    immediately on the closed queue. If drain timed out, workers are stuck+  --    mid-upload; interrupt them so the daemon process can exit promptly.+  withTakeMVar daemonWorkerThreads $+    if drained then Worker.stopWorkers else Worker.abortWorkers++  -- 6. Close all event subscriptions+  withTakeMVar daemonSubscriptionManagerThread (shutdownSubscriptions daemonSubscriptionManager)+   return pushResult   where     drainWithLogging daemonPushManager = do@@ -373,6 +388,7 @@       if drained         then Katip.logFM Katip.DebugS "Push manager drained."         else Katip.logFM Katip.WarningS "Push manager drain timed out. Some jobs may not have completed."+      pure drained      shutdownSubscriptions daemonSubscriptionManager subscriptionManagerThread = do       Katip.logFM Katip.DebugS "Shutting down event manager..."@@ -382,19 +398,55 @@      sayGoodbye socketStore exitResult socket = do       let clientSock = SocketStore.socket socket-      let clientThread = SocketStore.handlerThread socket-      let clientSocketId = SocketStore.socketId socket+          clientThread = SocketStore.handlerThread socket+          clientSocketId = SocketStore.socketId socket++      -- Stop the handler first so we have exclusive read access to the socket+      -- below. sendAll uses an independent sendLock, so the bye message still+      -- goes out cleanly.       Async.cancel clientThread -      -- Wave goodbye to the client that requested the shutdown-      liftIO $ Listen.serverBye clientSocketId socketStore 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."+      sendResult <-+        liftIO $+          tryAny $+            Listen.serverBye clientSocketId socketStore exitResult+      case sendResult of+        Left err ->+          Katip.logFM Katip.WarningS $+            Katip.ls $+              "Failed to send DaemonExit: " <> displayException err+        Right () -> pure ()++      -- Half-close the write side. Pending bytes (the bye message) flush to+      -- the client; the client reads them, then sees EOF.+      liftIO $ safeShutdown clientSock Socket.ShutdownSend++      -- Bounded wait for the client to read the bye message and close its end.+      mDisconnect <-+        liftIO $+          timeout disconnectTimeoutUs $+            try $+              Socket.BS.recv clientSock 4096+      case mDisconnect of+        Nothing ->+          Katip.logFM Katip.DebugS "Client did not disconnect within timeout."+        Just (Left err)+          | isResourceVanishedError err ->+              Katip.logFM Katip.DebugS "Client did not disconnect cleanly."+        Just (Left err) ->+          Katip.logFM Katip.DebugS $+            Katip.ls $+              "Client socket threw an error: " <> displayException err+        Just (Right _) ->+          Katip.logFM Katip.DebugS "Client disconnected."++      liftIO $ safeShutdown clientSock Socket.ShutdownBoth++    safeShutdown sock mode =+      Socket.shutdown sock mode `catchAny` \_ -> pure ()++    disconnectTimeoutUs :: Int+    disconnectTimeoutUs = 5_000_000  withTakeMVar :: (MonadUnliftIO m) => MVar a -> (a -> m ()) -> m () withTakeMVar mvar f = do
src/Cachix/Daemon/Listen.hs view
@@ -23,7 +23,6 @@   ) import Cachix.Daemon.Types.EventLoop (EventLoop) import Cachix.Daemon.Types.SocketStore (SocketId, SocketStore)-import Control.Exception.Safe (catchAny) import Control.Monad.Catch qualified as E import Data.Aeson qualified as Aeson import Data.ByteString qualified as BS@@ -117,7 +116,7 @@  serverBye :: SocketId -> SocketStore -> Either DaemonError () -> IO () serverBye socketId socketStore exitResult =-  SocketStore.sendAll socketId (Protocol.newMessage (DaemonExit exitStatus)) socketStore `catchAny` (\_ -> return ())+  SocketStore.sendAll socketId (Protocol.newMessage (DaemonExit exitStatus)) socketStore   where     exitStatus = DaemonExitStatus {exitCode, exitMessage}     exitCode = toExitCodeInt exitResult
src/Cachix/Daemon/Log.hs view
@@ -1,6 +1,8 @@ module Cachix.Daemon.Log   ( new,     withLogger,+    namespace,+    logMsg,     getKatipNamespace,     getKatipContext,     getKatipLogEnv,@@ -22,12 +24,22 @@ import Katip.Scribes.Handle (brackets, colorBySeverity, getKeys) import Protolude +-- | The default namespace for the Cachix daemon logger.+namespace :: Katip.Namespace+namespace = "cachix.daemon"+ new :: (MonadIO m) => Katip.Namespace -> Maybe Handle -> LogLevel -> m Logger new logLabel logHandle logLevel = do   logKLogEnv <- liftIO $ Katip.initLogEnv logLabel ""-  let logKNamespace = mempty+  let logKNamespace = logLabel   let logKContext = mempty   return $ Logger {..}++-- | Log a single message using a 'Logger' from outside the 'Katip' monad.+logMsg :: (MonadIO m) => Logger -> Katip.Severity -> Katip.LogStr -> m ()+logMsg logger sev msg =+  Katip.runKatipT (logKLogEnv logger) $+    Katip.logMsg (logKNamespace logger) sev msg  withLogger :: (MonadIO m, E.MonadMask m) => Logger -> (Logger -> m a) -> m a withLogger logger@(Logger {..}) f = do
src/Cachix/Daemon/Push.hs view
@@ -12,14 +12,16 @@   ( PushOptions (..),   ) import Cachix.Client.Push as Client.Push-import Cachix.Client.Retry (retryHttp)+import Cachix.Client.Retry (retryClientM) import Cachix.Client.Servant+import Cachix.Daemon.Log qualified as Log 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 Katip qualified import Protolude hiding (toS) import Servant.Auth () import Servant.Auth.Client@@ -51,12 +53,13 @@       pushParamsStore = store     } -getBinaryCache :: Env -> Maybe Token -> BinaryCacheName -> IO BinaryCache-getBinaryCache env authToken name = do+getBinaryCache :: Log.Logger -> Env -> Maybe Token -> BinaryCacheName -> IO BinaryCache+getBinaryCache logger env authToken name = do+  Log.logMsg logger Katip.DebugS $ "Fetching cache info for " <> Katip.ls name   -- 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+  res <- retryClientM (clientenv env) $ API.getCache cachixClient token name   case res of     Left err -> handleCacheResponse name authToken err     Right binaryCache -> pure binaryCache
src/Cachix/Daemon/PushManager.hs view
@@ -20,6 +20,7 @@     -- * Query     filterPushJobs,     getFailedPushJobs,+    failPendingJobs,      -- * Store paths     queueStorePaths,@@ -109,12 +110,14 @@ runPushManager :: (MonadIO m) => PushManagerEnv -> PushManager a -> m a runPushManager env f = liftIO $ unPushManager f `runReaderT` env --- | Set the shutdown latch (rejecting new jobs), then wait for all--- in-flight jobs to complete with a timeout.--- Returns True if all jobs completed, False if timed out.+-- | Set the shutdown latch (rejecting new jobs), then wait for all in-flight jobs to complete with an idle timeout.+--+-- Returns True if all jobs completed, False if no progress was observed within the timeout window. drainPushManager :: TimeoutOptions -> PushManagerEnv -> IO Bool drainPushManager timeoutOptions PushManagerEnv {..} = do   ShutdownLatch.initiateShutdown () pmShutdownLatch+  -- Reset the last event timestamp to start off the countdown from here+  updateTimestampTVar pmLastEventTimestamp   atomicallyWithTimeout timeoutOptions pmLastEventTimestamp $ do     pendingJobs <- readTVar pmPendingJobCount     check (pendingJobs <= 0)@@ -208,31 +211,70 @@     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)+-- | Apply an update to many push jobs atomically. After the update, a job+-- transitions to a terminal state if either the update made it processed+-- directly or it leaves 'pushQueue' empty. Returns the jobs that transitioned+-- to a terminal state in this call.+--+-- Mark, completion check, and pending-counter decrement happen in one STM+-- transaction so concurrent workers cannot race past the empty-queue check.+applyPushJobUpdates ::+  (Foldable t) =>+  t Protocol.PushRequestId ->+  (UTCTime -> PushJob -> PushJob) ->+  PushManager [PushJob]+applyPushJobUpdates pushIds update = do+  PushManagerEnv {pmPushJobs, pmPendingJobCount} <- ask+  ts <- liftIO getCurrentTime+  liftIO $ atomically $ do+    jobs <- readTVar pmPushJobs+    let (jobs', finished) = foldl' (step ts) (jobs, []) pushIds+    writeTVar pmPushJobs jobs'+    modifyTVar' pmPendingJobCount (subtract (length finished))+    pure finished+  where+    step ts (!jobs, acc) pushId =+      case HashMap.lookup pushId jobs of+        Just job+          | not (PushJob.isProcessed job) ->+              let job' = transitionIfDone ts (update ts job)+                  jobs' = HashMap.insert pushId job' jobs+               in if PushJob.isProcessed job'+                    then (jobs', job' : acc)+                    else (jobs', acc)+        _ -> (jobs, acc) -modifyPushJobs :: (Foldable f) => f Protocol.PushRequestId -> (PushJob -> PushJob) -> PushManager ()-modifyPushJobs pushIds f = do-  pushJobs <- asks pmPushJobs-  liftIO $ atomically $ modifyTVar' pushJobs $ \pushJobs' ->-    foldl' (flip (HashMap.adjust f)) pushJobs' pushIds+    transitionIfDone ts job+      | PushJob.isProcessed job = job+      | Set.null (PushJob.pushQueue job) =+          if PushJob.hasFailedPaths job+            then PushJob.fail ts job+            else PushJob.complete ts job+      | otherwise = job  failPushJob :: Protocol.PushRequestId -> PushManager ()-failPushJob pushId = do-  PushManagerEnv {..} <- ask-  timestamp <- liftIO getCurrentTime-  liftIO $ atomically $ do-    _ <- modifyPushJobSTM pmPushJobs pushId $ PushJob.fail timestamp-    decrementTVar pmPendingJobCount+failPushJob pushId = void $ applyPushJobUpdates [pushId] PushJob.fail +-- | Mark every non-terminal job as failed and emit failure events for any+-- paths still in their queues. Failed jobs stay in 'pmPushJobs' so a later+-- 'getFailedPushJobs' call sees them and the daemon exits with the right+-- code; the natural path removes jobs via 'pushFinished'.+failPendingJobs :: Text -> PushManager [PushJob]+failPendingJobs reason = do+  pmPushJobs <- asks pmPushJobs+  allIds <- HashMap.keys <$> liftIO (readTVarIO pmPushJobs)+  failed <- applyPushJobUpdates allIds PushJob.fail++  ts <- liftIO getCurrentTime+  sendPushEvent <- asks pmOnPushEvent+  for_ failed $ \job -> do+    let pid = PushJob.pushId job+    for_ (PushJob.pushQueue job) $ \path ->+      sendStorePathEventAt ts [pid] (PushStorePathFailed path reason)+    liftIO $ sendPushEvent pid (PushEvent ts pid PushFinished)++  pure failed+ pendingJobCount :: PushManager Int pendingJobCount = do   pmPendingJobCount <- asks pmPendingJobCount@@ -265,21 +307,6 @@   references <- liftIO $ readTVarIO storePathIndex   return $ fromMaybe Seq.empty (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@@ -292,19 +319,17 @@ resolvePushJob pushId closure = do   Katip.logLocM Katip.DebugS $ Katip.ls $ showClosureStats closure -  timestamp <- liftIO getCurrentTime-  _ <- modifyPushJob pushId $ PushJob.populateQueue closure timestamp+  finishedJobs <- applyPushJobUpdates [pushId] (PushJob.populateQueue closure)    withPushJob pushId $ \pushJob -> do     pushStarted pushJob-    -- Emit PushStorePathSkipped events for paths already present in the cache     let skippedPaths = Set.difference (PushJob.rcAllPaths closure) (PushJob.rcMissingPaths closure)+    ts <- liftIO getCurrentTime     forM_ skippedPaths $ \path ->-      sendStorePathEvent [pushId] (PushStorePathSkipped path)-    -- Create STM action for each path and then run everything atomically+      sendStorePathEventAt ts [pushId] (PushStorePathSkipped path)     queueStorePaths pushId $ Set.toList (PushJob.rcMissingPaths closure)-    -- Check if the job is already completed, i.e. all paths have been skipped.-    checkPushJobCompleted pushId++  for_ finishedJobs pushFinished   where     showClosureStats :: PushJob.ResolvedClosure FilePath -> Text     showClosureStats PushJob.ResolvedClosure {..} =@@ -358,8 +383,9 @@         liftIO $ for_ errors $ uncurry logStorePathWarning          -- Emit PushStorePathInvalid events for invalid paths+        ts <- liftIO getCurrentTime         forM_ errors $ \(path, err) ->-          sendStorePathEvent [pushId] (PushStorePathInvalid path (formatStorePathError err))+          sendStorePathEventAt ts [pushId] (PushStorePathInvalid path (formatStorePathError err))          paths <- computeClosure store validPaths @@ -532,6 +558,10 @@ sendStorePathEvent :: (Foldable f) => f Protocol.PushRequestId -> PushEventMessage -> PushManager () sendStorePathEvent pushIds msg = do   timestamp <- liftIO getCurrentTime+  sendStorePathEventAt timestamp pushIds msg++sendStorePathEventAt :: (Foldable f) => UTCTime -> f Protocol.PushRequestId -> PushEventMessage -> PushManager ()+sendStorePathEventAt timestamp pushIds msg = do   sendPushEvent <- asks pmOnPushEvent   liftIO $ forM_ pushIds $ \pushId ->     sendPushEvent pushId (PushEvent timestamp pushId msg)@@ -550,23 +580,17 @@ pushStorePathDone :: FilePath -> PushManager () pushStorePathDone storePath = do   pushIds <- lookupStorePathIndex storePath-  modifyPushJobs pushIds (PushJob.markStorePathPushed storePath)-+  finishedJobs <- applyPushJobUpdates pushIds (\_ -> PushJob.markStorePathPushed storePath)   sendStorePathEvent pushIds (PushStorePathDone storePath)--  mapM_ checkPushJobCompleted pushIds-+  for_ finishedJobs pushFinished   removeStorePath storePath  pushStorePathFailed :: FilePath -> Text -> PushManager () pushStorePathFailed storePath errMsg = do   pushIds <- lookupStorePathIndex storePath-  modifyPushJobs pushIds (PushJob.markStorePathFailed storePath)-+  finishedJobs <- applyPushJobUpdates pushIds (\_ -> PushJob.markStorePathFailed storePath)   sendStorePathEvent pushIds (PushStorePathFailed storePath errMsg)--  mapM_ checkPushJobCompleted pushIds-+  for_ finishedJobs pushFinished   removeStorePath storePath  -- Helpers
src/Cachix/Daemon/Types/Daemon.hs view
@@ -105,11 +105,12 @@   localKatipNamespace f (Daemon m) = Daemon (local (\s -> s {daemonLogger = Log.localKatipNamespace f (daemonLogger s)}) m)  -- | Run a pre-configured daemon.+--+-- The caller is responsible for setting up logging (e.g. via 'Log.withLogger')+-- before calling this function. The 'daemonLogger' in the environment must+-- already have its scribes registered. 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}+runDaemon env f = tryDaemon $ unDaemon f `runReaderT` env   where     tryDaemon :: IO a -> IO (Either DaemonError a)     tryDaemon a =
src/Cachix/Daemon/Worker.hs view
@@ -1,6 +1,7 @@ module Cachix.Daemon.Worker   ( startWorkers,     stopWorkers,+    abortWorkers,     startWorker,     stopWorker,     Immortal.Thread,@@ -43,6 +44,12 @@ stopWorker worker = liftIO $ do   Immortal.mortalize worker   Immortal.wait worker++-- | Stop workers by interrupting their current task with an async exception.+abortWorkers :: (Katip.KatipContext m) => [Immortal.Thread] -> m ()+abortWorkers workers = do+  Katip.logFM Katip.WarningS "Aborting workers."+  liftIO $ Async.mapConcurrently_ Immortal.stop workers  logWorkerException :: (Exception e, Katip.KatipContext m) => Either e () -> m () logWorkerException (Left err) =