cachix 1.7 → 1.7.1
raw patch · 6 files changed
+77/−27 lines, 6 files
Files
- CHANGELOG.md +8/−0
- cachix.cabal +1/−1
- src/Cachix/Client/Daemon.hs +18/−6
- src/Cachix/Client/Daemon/Listen.hs +15/−4
- src/Cachix/Client/Daemon/PushManager.hs +22/−16
- src/Cachix/Client/Daemon/PushManager/PushJob.hs +13/−0
CHANGELOG.md view
@@ -5,6 +5,14 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [1.7.1] - 2023-02-20++## Fixed++- daemon: add explicit sigINT/sigTERM handler+- daemon: improve shutdown when jobs fail+- daemon: log decoding errors+ ## [1.7] - 2023-01-08 ## Added
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix-version: 1.7+version: 1.7.1 synopsis: Command-line client for Nix binary cache hosting https://cachix.org
src/Cachix/Client/Daemon.hs view
@@ -33,10 +33,12 @@ import Control.Exception.Safe (catchAny) import qualified Control.Monad.Catch as E import qualified Data.Text as T+import qualified Hercules.CNix.Util as CNix.Util import qualified Katip import qualified Network.Socket as Socket import Protolude import System.Posix.Process (getProcessID)+import qualified System.Posix.Signals as Signal import qualified UnliftIO.Async as Async -- | Configure a new daemon. Use 'run' to start it.@@ -78,6 +80,7 @@ start :: Env -> DaemonOptions -> PushOptions -> BinaryCacheName -> IO () start daemonEnv daemonOptions daemonPushOptions daemonCacheName = do daemon <- new daemonEnv daemonOptions Nothing daemonPushOptions daemonCacheName+ installSignalHandlers daemon void $ run daemon -- | Run a daemon from a given configuration@@ -99,19 +102,17 @@ subscriptionManagerThread <- liftIO $ Async.async $ runSubscriptionManager daemonSubscriptionManager - let shutdownQueue =+ let stopPushManager = liftIO $ PushManager.stopPushManager daemonPushManager Push.withPushParams $ \pushParams -> E.bracketOnError (startWorkers pushParams) Worker.stopWorkers $ \workers -> do- flip E.onError shutdownQueue $+ flip E.onError stopPushManager $ -- TODO: retry the connection on socket errors E.bracketOnError (Daemon.openSocket daemonSocketPath) Daemon.closeSocket $ \sock -> do liftIO $ Socket.listen sock Socket.maxListenQueue - res <-- Async.race (waitForShutdown daemonShutdownLatch) $- Daemon.listen queueJob sock `E.finally` stop+ listenThread <- Async.async $ Daemon.listen stop queueJob sock waitForShutdown daemonShutdownLatch @@ -126,7 +127,7 @@ -- Stop receiving new push requests liftIO $ Socket.shutdown sock Socket.ShutdownReceive `catchAny` \_ -> return () - shutdownQueue+ stopPushManager -- Gracefully shutdown the worker *before* closing the socket Worker.stopWorkers workers@@ -135,6 +136,8 @@ Async.wait subscriptionManagerThread -- TODO: say goodbye to all clients waiting for their push to go through+ Async.cancel listenThread+ res <- Async.waitCatch listenThread case res of Right clientSock -> do -- Wave goodbye to the client that requested the shutdown@@ -150,6 +153,15 @@ stopIO :: DaemonEnv -> IO () stopIO DaemonEnv {daemonShutdownLatch} = initiateShutdown daemonShutdownLatch++installSignalHandlers :: DaemonEnv -> IO ()+installSignalHandlers daemon = do+ for_ [Signal.sigTERM, Signal.sigINT] $ \signal ->+ Signal.installHandler signal (Signal.CatchOnce handler) Nothing+ where+ handler = do+ CNix.Util.triggerInterrupt+ stopIO daemon queueJob :: Protocol.PushRequest -> Socket.Socket -> Daemon () queueJob pushRequest _clientConn = do
src/Cachix/Client/Daemon/Listen.hs view
@@ -12,6 +12,7 @@ import Control.Exception.Safe (catchAny) import qualified Control.Exception.Safe as Safe import qualified Data.Aeson as Aeson+import qualified Katip import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.BS import qualified Network.Socket.ByteString.Lazy as Socket.LBS@@ -37,20 +38,30 @@ DecodingError err -> "Failed to decode request: " <> toS err -- | The main daemon server loop.-listen :: (MonadIO m) => (Protocol.PushRequest -> Socket.Socket -> m ()) -> Socket.Socket -> m Socket.Socket-listen pushToQueue sock = loop+listen ::+ (Katip.KatipContext m) =>+ -- | An action to run when the daemon is requested to stop+ m () ->+ -- | An action to run when a push request is received+ (Protocol.PushRequest -> Socket.Socket -> m ()) ->+ -- | An active socket to listen on+ Socket.Socket ->+ -- | Returns a socket to the client that requested the daemon to stop+ m Socket.Socket+listen onClientStop onPushRequest sock = loop where loop = do result <- liftIO $ runExceptT $ readPushRequest sock case result of Right (ClientStop, clientConn) -> do+ onClientStop return clientConn Right (ClientPushRequest pushRequest, clientConn) -> do- pushToQueue pushRequest clientConn+ onPushRequest pushRequest clientConn loop Left err@(DecodingError _) -> do- putErrText $ toS $ displayException err+ Katip.logFM Katip.ErrorS $ Katip.ls $ displayException err loop Left err -> throwIO err
src/Cachix/Client/Daemon/PushManager.hs view
@@ -81,7 +81,7 @@ stopPushManager PushManagerEnv {pmTaskQueue, pmPushJobs} = atomically $ do pushJobs <- readTVar pmPushJobs- if all PushJob.isCompleted (HashMap.elems pushJobs)+ if all PushJob.isProcessed (HashMap.elems pushJobs) then closeTBMQueue pmTaskQueue else retry @@ -131,6 +131,11 @@ liftIO $ atomically $ modifyTVar' pushJobs $ \pushJobs' -> foldl' (flip (HashMap.adjust f)) pushJobs' pushIds +failPushJob :: Protocol.PushRequestId -> PushManager ()+failPushJob pushId = do+ timestamp <- liftIO getCurrentTime+ void $ modifyPushJob pushId $ PushJob.fail timestamp+ -- Manage store paths queueStorePaths :: Protocol.PushRequestId -> [FilePath] -> PushManager ()@@ -210,23 +215,24 @@ ResolveClosure pushId -> do Katip.logLocM Katip.DebugS $ Katip.ls $ "Resolving closure for push job " <> (show pushId :: Text) - withPushJob pushId $ \pushJob -> do- let sps = Protocol.storePaths (pushRequest pushJob)- store = pushParamsStore pushParams- normalized <- mapM (normalizeStorePath store) sps- (allStorePaths, missingStorePaths) <- getMissingPathsForClosure pushParams (catMaybes normalized)- storePathsToPush <- pushOnClosureAttempt pushParams allStorePaths missingStorePaths+ withPushJob pushId $ \pushJob ->+ E.onException (failPushJob pushId) $ do+ let sps = Protocol.storePaths (pushRequest pushJob)+ store = pushParamsStore pushParams+ normalized <- mapM (normalizeStorePath store) sps+ (allStorePaths, missingStorePaths) <- getMissingPathsForClosure pushParams (catMaybes normalized)+ storePathsToPush <- pushOnClosureAttempt pushParams allStorePaths missingStorePaths - resolvedClosure <- do- allPaths <- liftIO $ mapM (storeToFilePath store) allStorePaths- pathsToPush <- liftIO $ mapM (storeToFilePath store) storePathsToPush- return $- PushJob.ResolvedClosure- { rcAllPaths = Set.fromList allPaths,- rcMissingPaths = Set.fromList pathsToPush- }+ resolvedClosure <- do+ allPaths <- liftIO $ mapM (storeToFilePath store) allStorePaths+ pathsToPush <- liftIO $ mapM (storeToFilePath store) storePathsToPush+ return $+ PushJob.ResolvedClosure+ { rcAllPaths = Set.fromList allPaths,+ rcMissingPaths = Set.fromList pathsToPush+ } - resolvePushJob pushId resolvedClosure+ resolvePushJob pushId resolvedClosure PushStorePath filePath -> do qs <- asks pmTaskSemaphore E.bracket_ (QSem.waitQSem qs) (QSem.signalQSem qs) $ do
src/Cachix/Client/Daemon/PushManager/PushJob.hs view
@@ -90,8 +90,21 @@ pushStats = pushStats {jsCompletedAt = Just timestamp} } +fail :: UTCTime -> PushJob -> PushJob+fail timestamp pushJob@PushJob {..} = do+ pushJob+ { pushStatus = Failed,+ pushStats = pushStats {jsCompletedAt = Just timestamp}+ }+ isCompleted :: PushJob -> Bool isCompleted PushJob {pushStatus} = pushStatus == Completed++isFailed :: PushJob -> Bool+isFailed PushJob {pushStatus} = pushStatus == Failed++isProcessed :: PushJob -> Bool+isProcessed pushJob = isCompleted pushJob || isFailed pushJob startedAt :: PushJob -> Maybe UTCTime startedAt PushJob {pushStats = JobStats {jsStartedAt}} = jsStartedAt