diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/cachix.cabal b/cachix.cabal --- a/cachix.cabal +++ b/cachix.cabal @@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix -version: 1.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 diff --git a/src/Cachix/Client/Command/Cache.hs b/src/Cachix/Client/Command/Cache.hs --- a/src/Cachix/Client/Command/Cache.hs +++ b/src/Cachix/Client/Command/Cache.hs @@ -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 diff --git a/src/Cachix/Client/Command/Config.hs b/src/Cachix/Client/Command/Config.hs --- a/src/Cachix/Client/Command/Config.hs +++ b/src/Cachix/Client/Command/Config.hs @@ -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] diff --git a/src/Cachix/Client/Command/Doctor.hs b/src/Cachix/Client/Command/Doctor.hs --- a/src/Cachix/Client/Command/Doctor.hs +++ b/src/Cachix/Client/Command/Doctor.hs @@ -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 diff --git a/src/Cachix/Client/Command/Pin.hs b/src/Cachix/Client/Command/Pin.hs --- a/src/Cachix/Client/Command/Pin.hs +++ b/src/Cachix/Client/Command/Pin.hs @@ -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 diff --git a/src/Cachix/Client/Command/Push.hs b/src/Cachix/Client/Command/Push.hs --- a/src/Cachix/Client/Command/Push.hs +++ b/src/Cachix/Client/Command/Push.hs @@ -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) diff --git a/src/Cachix/Client/Command/Watch.hs b/src/Cachix/Client/Command/Watch.hs --- a/src/Cachix/Client/Command/Watch.hs +++ b/src/Cachix/Client/Command/Watch.hs @@ -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 diff --git a/src/Cachix/Client/Push.hs b/src/Cachix/Client/Push.hs --- a/src/Cachix/Client/Push.hs +++ b/src/Cachix/Client/Push.hs @@ -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 diff --git a/src/Cachix/Client/Push/S3.hs b/src/Cachix/Client/Push/S3.hs --- a/src/Cachix/Client/Push/S3.hs +++ b/src/Cachix/Client/Push/S3.hs @@ -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 diff --git a/src/Cachix/Client/Retry.hs b/src/Cachix/Client/Retry.hs --- a/src/Cachix/Client/Retry.hs +++ b/src/Cachix/Client/Retry.hs @@ -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 diff --git a/src/Cachix/Daemon.hs b/src/Cachix/Daemon.hs --- a/src/Cachix/Daemon.hs +++ b/src/Cachix/Daemon.hs @@ -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 diff --git a/src/Cachix/Daemon/Listen.hs b/src/Cachix/Daemon/Listen.hs --- a/src/Cachix/Daemon/Listen.hs +++ b/src/Cachix/Daemon/Listen.hs @@ -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 diff --git a/src/Cachix/Daemon/Log.hs b/src/Cachix/Daemon/Log.hs --- a/src/Cachix/Daemon/Log.hs +++ b/src/Cachix/Daemon/Log.hs @@ -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 diff --git a/src/Cachix/Daemon/Push.hs b/src/Cachix/Daemon/Push.hs --- a/src/Cachix/Daemon/Push.hs +++ b/src/Cachix/Daemon/Push.hs @@ -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 diff --git a/src/Cachix/Daemon/PushManager.hs b/src/Cachix/Daemon/PushManager.hs --- a/src/Cachix/Daemon/PushManager.hs +++ b/src/Cachix/Daemon/PushManager.hs @@ -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 diff --git a/src/Cachix/Daemon/Types/Daemon.hs b/src/Cachix/Daemon/Types/Daemon.hs --- a/src/Cachix/Daemon/Types/Daemon.hs +++ b/src/Cachix/Daemon/Types/Daemon.hs @@ -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 = diff --git a/src/Cachix/Daemon/Worker.hs b/src/Cachix/Daemon/Worker.hs --- a/src/Cachix/Daemon/Worker.hs +++ b/src/Cachix/Daemon/Worker.hs @@ -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) =