cachix 1.7.2 → 1.7.3
raw patch · 23 files changed
+729/−192 lines, 23 files
Files
- CHANGELOG.md +7/−0
- cachix-deployment/Main.hs +5/−3
- cachix.cabal +8/−1
- src/Cachix/Client/Commands.hs +5/−5
- src/Cachix/Client/Daemon.hs +95/−54
- src/Cachix/Client/Daemon/Client.hs +87/−20
- src/Cachix/Client/Daemon/EventLoop.hs +73/−0
- src/Cachix/Client/Daemon/Listen.hs +63/−46
- src/Cachix/Client/Daemon/PostBuildHook.hs +40/−6
- src/Cachix/Client/Daemon/Protocol.hs +6/−4
- src/Cachix/Client/Daemon/PushManager.hs +87/−16
- src/Cachix/Client/Daemon/PushManager/PushJob.hs +2/−4
- src/Cachix/Client/Daemon/SocketStore.hs +43/−0
- src/Cachix/Client/Daemon/Types/Daemon.hs +8/−0
- src/Cachix/Client/Daemon/Types/EventLoop.hs +35/−0
- src/Cachix/Client/Daemon/Types/PushEvent.hs +2/−1
- src/Cachix/Client/Daemon/Types/PushManager.hs +14/−0
- src/Cachix/Client/Daemon/Types/SocketStore.hs +23/−0
- src/Cachix/Client/Daemon/Worker.hs +1/−0
- src/Cachix/Client/NixVersion.hs +46/−15
- src/Cachix/Deploy/Lock.hs +6/−0
- test/Daemon/PushManagerSpec.hs +40/−0
- test/NixVersionSpec.hs +33/−17
CHANGELOG.md view
@@ -5,6 +5,13 @@ 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.3] - 2024-05-17++- daemon: implement robust shutdown handling and fix occasional hangs+- daemon: respect RUNNER_TEMP+- lix: support parsing lix version strings+- deploy: fix path to lock file+ ## [1.7.2] - 2024-03-06 ### Added
cachix-deployment/Main.hs view
@@ -31,8 +31,8 @@ import Protolude.Conv import System.IO (BufferMode (..), hSetBuffering) -lockFilename :: Text -> FilePath-lockFilename agentName = "deployment-" <> toS agentName+lockFilenameFrom :: Text -> FilePath+lockFilenameFrom agentName = "deployment-" <> toS agentName -- | Activate the new deployment. --@@ -76,8 +76,10 @@ WebSocket.identifier = Agent.agentIdentifier agentName } + lockFilePath <- Lock.newLockFilePath (lockFilenameFrom agentName)+ Log.withLog logOptions $ \withLog ->- void . Lock.withTryLock (lockFilename agentName) $ do+ void . Lock.withTryLock lockFilePath $ do -- Open a connection to logging stream (logQueue, loggingThread) <- runLogStream withLog logWebsocketOptions
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: cachix-version: 1.7.2+version: 1.7.3 synopsis: Command-line client for Nix binary cache hosting https://cachix.org @@ -69,6 +69,7 @@ Cachix.Client.Config.Orphans Cachix.Client.Daemon Cachix.Client.Daemon.Client+ Cachix.Client.Daemon.EventLoop Cachix.Client.Daemon.Listen Cachix.Client.Daemon.Log Cachix.Client.Daemon.PostBuildHook@@ -78,12 +79,15 @@ Cachix.Client.Daemon.PushManager Cachix.Client.Daemon.PushManager.PushJob Cachix.Client.Daemon.ShutdownLatch+ Cachix.Client.Daemon.SocketStore Cachix.Client.Daemon.Subscription Cachix.Client.Daemon.Types Cachix.Client.Daemon.Types.Daemon+ Cachix.Client.Daemon.Types.EventLoop Cachix.Client.Daemon.Types.Log Cachix.Client.Daemon.Types.PushEvent Cachix.Client.Daemon.Types.PushManager+ Cachix.Client.Daemon.Types.SocketStore Cachix.Client.Daemon.Worker Cachix.Client.Env Cachix.Client.Exception@@ -249,6 +253,7 @@ build-depends: , aeson+ , async , bytestring , cachix , cachix-api@@ -262,7 +267,9 @@ , retry , servant-auth-client , servant-client-core+ , stm , temporary , time+ , versions build-tool-depends: hspec-discover:hspec-discover
src/Cachix/Client/Commands.hs view
@@ -93,7 +93,7 @@ import System.Environment (getEnvironment) import System.IO (hIsTerminalDevice) import System.IO.Error (isEOFError)-import System.IO.Temp (withSystemTempFile)+import System.IO.Temp (withTempFile) import qualified System.Posix.Signals as Signals import qualified System.Process import qualified URI.ByteString as UBS@@ -376,15 +376,15 @@ -- Requires the user to be a trusted user in a multi-user installation. watchExecDaemon :: Env -> PushOptions -> BinaryCacheName -> Text -> [Text] -> IO () watchExecDaemon env pushOpts cacheName cmd args =- Daemon.PostBuildHook.withSetup Nothing $ \daemonSock nixConfEnv ->- withSystemTempFile "daemon-log-capture" $ \_ logHandle -> do- let daemonOptions = DaemonOptions {daemonSocketPath = Just daemonSock}+ Daemon.PostBuildHook.withSetup Nothing $ \hookEnv ->+ withTempFile (Daemon.PostBuildHook.tempDir hookEnv) "daemon-log-capture" $ \_ logHandle -> do+ let daemonOptions = DaemonOptions {daemonSocketPath = Just (Daemon.PostBuildHook.daemonSock hookEnv)} daemon <- Daemon.new env daemonOptions (Just logHandle) pushOpts cacheName exitCode <- bracket (startDaemonThread daemon) (shutdownDaemonThread daemon logHandle) $ \_ -> do processEnv <- getEnvironment- let newProcessEnv = Daemon.PostBuildHook.modifyEnv nixConfEnv processEnv+ let newProcessEnv = Daemon.PostBuildHook.modifyEnv (Daemon.PostBuildHook.envVar hookEnv) processEnv let process = (System.Process.proc (toS cmd) (toS <$> args)) { System.Process.std_out = System.Process.Inherit,
src/Cachix/Client/Daemon.hs view
@@ -13,14 +13,17 @@ import qualified Cachix.Client.Commands.Push as Commands.Push import qualified Cachix.Client.Config as Config import Cachix.Client.Config.Orphans ()+import qualified Cachix.Client.Daemon.EventLoop as EventLoop import Cachix.Client.Daemon.Listen as Daemon import qualified Cachix.Client.Daemon.Log as Log import Cachix.Client.Daemon.Protocol as Protocol import Cachix.Client.Daemon.Push as Push import qualified Cachix.Client.Daemon.PushManager as PushManager import Cachix.Client.Daemon.ShutdownLatch+import qualified Cachix.Client.Daemon.SocketStore as SocketStore import Cachix.Client.Daemon.Subscription as Subscription import Cachix.Client.Daemon.Types as Types+import Cachix.Client.Daemon.Types.EventLoop (DaemonEvent (ShutdownGracefully)) import qualified Cachix.Client.Daemon.Types.PushManager as PushManager import qualified Cachix.Client.Daemon.Worker as Worker import Cachix.Client.Env as Env@@ -36,7 +39,9 @@ import qualified Hercules.CNix.Util as CNix.Util import qualified Katip import qualified Network.Socket as Socket+import qualified Network.Socket.ByteString as Socket.BS import Protolude+import System.IO.Error (isResourceVanishedError) import System.Posix.Process (getProcessID) import qualified System.Posix.Signals as Signal import qualified UnliftIO.Async as Async@@ -63,6 +68,9 @@ daemonLogger <- Log.new "cachix.daemon" daemonLogHandle daemonLogLevel daemonSocketPath <- maybe getSocketPath pure (Options.daemonSocketPath daemonOptions)+ daemonSocketThread <- newEmptyMVar+ daemonEventLoop <- EventLoop.new+ daemonClients <- SocketStore.newSocketStore daemonShutdownLatch <- newShutdownLatch daemonPid <- getProcessID @@ -83,18 +91,17 @@ installSignalHandlers daemon void $ run daemon --- | Run a daemon from a given configuration+-- | Run a daemon from a given configuration. run :: DaemonEnv -> IO ExitCode run daemon = runDaemon daemon $ flip E.onError (return $ ExitFailure 1) $ do Katip.logFM Katip.InfoS "Starting Cachix Daemon" DaemonEnv {..} <- ask - config <- showConfiguration- Katip.logFM Katip.InfoS $ Katip.ls $ "Configuration:\n" <> config+ printConfiguration Push.withPushParams $ \pushParams -> do subscriptionManagerThread <-- liftIO $ Async.async $ runSubscriptionManager daemonSubscriptionManager+ Async.async $ runSubscriptionManager daemonSubscriptionManager let runWorkerTask = liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask pushParams@@ -104,56 +111,36 @@ (PushManager.pmTaskQueue daemonPushManager) runWorkerTask - -- TODO: retry the connection on socket errors- E.bracketOnError (Daemon.openSocket daemonSocketPath) Daemon.closeSocket $ \sock -> do- liftIO $ Socket.listen sock Socket.maxListenQueue- listenThread <- Async.async $ Daemon.listen stop queueJob sock-- -- Wait for a shutdown signal- waitForShutdown daemonShutdownLatch-- Katip.logFM Katip.InfoS "Shutting down daemon..."-- -- Stop receiving new push requests- liftIO $ Socket.shutdown sock Socket.ShutdownReceive `catchAny` \_ -> return ()-- PushManager.runPushManager daemonPushManager $ do- queuedStorePathCount <- PushManager.queuedStorePathCount- when (queuedStorePathCount > 0) $- Katip.logFM Katip.InfoS $- Katip.logStr $- "Remaining store paths: " <> (show queuedStorePathCount :: Text)-- -- Finish processing remaining push jobs- liftIO $ PushManager.stopPushManager daemonPushManager-- -- Gracefully shut down the worker before closing the socket- Worker.stopWorkers workersThreads-- -- Close all event subscriptions- liftIO $ stopSubscriptionManager daemonSubscriptionManager- Async.wait subscriptionManagerThread-- -- TODO: say goodbye to all clients waiting for their push to go through- listenThreadRes <- do- Async.cancel listenThread- Async.waitCatch listenThread-- case listenThreadRes of- Right clientSock -> do- -- Wave goodbye to the client that requested the shutdown- liftIO $ Daemon.serverBye clientSock- liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` \_ -> return ()- _ -> return ()+ listenThread <- Async.async (Daemon.listen daemonEventLoop daemonSocketPath)+ liftIO $ putMVar daemonSocketThread listenThread - return ExitSuccess+ EventLoop.run daemonEventLoop $ \case+ EventLoop.AddSocketClient conn ->+ SocketStore.addSocket conn (Daemon.handleClient daemonEventLoop) daemonClients+ EventLoop.RemoveSocketClient socketId ->+ SocketStore.removeSocket socketId daemonClients+ EventLoop.ReconnectSocket ->+ -- TODO: implement reconnection logic+ EventLoop.exitLoopWith (ExitFailure 1) daemonEventLoop+ EventLoop.ReceivedMessage clientMsg ->+ case clientMsg of+ ClientPushRequest pushRequest -> queueJob pushRequest+ ClientStop -> EventLoop.send daemonEventLoop EventLoop.ShutdownGracefully+ _ -> return ()+ EventLoop.ShutdownGracefully -> do+ Katip.logFM Katip.InfoS "Shutting down daemon..."+ shutdownGracefully subscriptionManagerThread workersThreads+ Katip.logFM Katip.InfoS "Daemon shut down. Exiting."+ EventLoop.exitLoopWith ExitSuccess daemonEventLoop stop :: Daemon ()-stop = asks daemonShutdownLatch >>= initiateShutdown+stop = do+ eventloop <- asks daemonEventLoop+ EventLoop.send eventloop ShutdownGracefully stopIO :: DaemonEnv -> IO ()-stopIO DaemonEnv {daemonShutdownLatch} =- initiateShutdown daemonShutdownLatch+stopIO DaemonEnv {daemonEventLoop} =+ EventLoop.sendIO daemonEventLoop ShutdownGracefully installSignalHandlers :: DaemonEnv -> IO () installSignalHandlers daemon = do@@ -164,10 +151,10 @@ CNix.Util.triggerInterrupt stopIO daemon -queueJob :: Protocol.PushRequest -> Socket.Socket -> Daemon ()-queueJob pushRequest _clientConn = do- DaemonEnv {..} <- ask- -- TODO: subscribe the socket to updates if requested+queueJob :: Protocol.PushRequest -> Daemon ()+queueJob pushRequest = do+ daemonPushManager <- asks daemonPushManager+ -- TODO: subscribe the socket to updates if available -- Queue the job void $@@ -180,7 +167,13 @@ subscribeToAllSTM daemonSubscriptionManager (SubChannel chan) dupTMChan chan --- | Print debug information about the daemon configuration+-- | Print the daemon configuration to the log.+printConfiguration :: Daemon ()+printConfiguration = do+ config <- showConfiguration+ Katip.logFM Katip.InfoS $ Katip.ls $ "Configuration:\n" <> config++-- | Fetch debug information about the daemon configuration. showConfiguration :: Daemon Text showConfiguration = do DaemonEnv {..} <- ask@@ -196,3 +189,51 @@ "Cache is public: " <> show (BinaryCache.isPublic daemonBinaryCache), "Compression: " <> show (Push.getCompressionMethod daemonPushOptions daemonBinaryCache) ]++shutdownGracefully :: Async () -> [Worker.Thread] -> Daemon ()+shutdownGracefully subscriptionManagerThread workersThreads = do+ DaemonEnv {..} <- ask++ initiateShutdown daemonShutdownLatch+ PushManager.runPushManager daemonPushManager $ do+ queuedStorePathCount <- PushManager.queuedStorePathCount+ when (queuedStorePathCount > 0) $+ Katip.logFM Katip.InfoS $+ Katip.logStr $+ "Remaining store paths: " <> (show queuedStorePathCount :: Text)++ Katip.logFM Katip.DebugS "Waiting for push manager to clear remaining jobs..."+ -- Finish processing remaining push jobs+ let timeoutOptions =+ PushManager.TimeoutOptions+ { PushManager.toTimeout = 60.0,+ PushManager.toPollingInterval = 1.0+ }+ liftIO $ PushManager.stopPushManager timeoutOptions daemonPushManager+ Katip.logFM Katip.DebugS "Push manager shut down."++ -- Gracefully shut down the worker before closing the socket+ Worker.stopWorkers workersThreads++ -- Close all event subscriptions+ Katip.logFM Katip.DebugS "Shutting down event manager..."+ liftIO $ stopSubscriptionManager daemonSubscriptionManager+ Async.wait subscriptionManagerThread+ Katip.logFM Katip.DebugS "Event manager shut down."++ let sayGoodbye socket = do+ let clientSock = SocketStore.socket socket+ let clientThread = SocketStore.handlerThread socket+ Async.cancel clientThread++ -- Wave goodbye to the client that requested the shutdown+ liftIO $ Daemon.serverBye clientSock+ liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` (\_ -> return ())+ -- Wait for the other end to disconnect+ ebs <- liftIO $ try $ Socket.BS.recv clientSock 4096+ case ebs of+ Left err | isResourceVanishedError err -> Katip.logFM Katip.DebugS "Client did not disconnect cleanly."+ Left err -> Katip.logFM Katip.DebugS $ Katip.ls $ "Client socket threw an error: " <> displayException err+ Right _ -> Katip.logFM Katip.DebugS "Client disconnected."++ Async.mapConcurrently_ sayGoodbye =<< SocketStore.toList daemonClients
src/Cachix/Client/Daemon/Client.hs view
@@ -4,23 +4,42 @@ import Cachix.Client.Daemon.Protocol as Protocol import Cachix.Client.Env as Env import Cachix.Client.OptionsParser (DaemonOptions (..))+import qualified Cachix.Client.Retry as Retry import qualified Control.Concurrent.Async as Async+import Control.Concurrent.STM.TBMQueue import Control.Exception.Safe (catchAny) import qualified Data.Aeson as Aeson import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy.Char8 as Lazy.Char8+import Data.IORef+import Data.Time.Clock import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.BS import qualified Network.Socket.ByteString.Lazy as Socket.LBS import Protolude-import qualified System.Posix.IO as Posix +data SocketError+ = -- | The socket has been closed+ SocketClosed+ | -- | The socket has stopped responding to pings+ SocketStalled+ | -- | Failed to decode a message from the socket+ SocketDecodingError !Text+ deriving stock (Show)++instance Exception SocketError where+ displayException = \case+ SocketClosed -> "The socket has been closed"+ SocketStalled -> "The socket has stopped responding to pings"+ SocketDecodingError err -> "Failed to decode message from socket: " <> toS err+ -- | Queue up push requests with the daemon -- -- TODO: wait for the daemon to respond that it has received the request push :: Env -> DaemonOptions -> [FilePath] -> IO () push _env daemonOptions storePaths = withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do- Socket.LBS.sendAll sock (Aeson.encode pushRequest)+ Socket.LBS.sendAll sock $ Aeson.encode pushRequest `Lazy.Char8.snoc` '\n' where pushRequest = Protocol.ClientPushRequest $@@ -30,36 +49,84 @@ stop :: Env -> DaemonOptions -> IO () stop _env daemonOptions = withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do- Async.concurrently_ (waitForResponse sock) $- Socket.LBS.sendAll sock (Aeson.encode Protocol.ClientStop)+ let size = 100+ (rx, tx) <- atomically $ (,) <$> newTBMQueue size <*> newTBMQueue size++ rxThread <- Async.async (handleIncoming rx sock)+ txThread <- Async.async (handleOutgoing tx sock)++ lastPongRef <- newIORef =<< getCurrentTime+ pingThread <- Async.async (runPingThread lastPongRef rx tx)++ mapM_ Async.link [rxThread, txThread, pingThread]++ -- Request the daemon to stop+ atomically $ writeTBMQueue tx Protocol.ClientStop++ fix $ \loop -> do+ mmsg <- atomically (readTBMQueue rx)+ case mmsg of+ Nothing -> return ()+ Just (Left err) -> putErrText $ toS $ displayException err+ Just (Right msg) ->+ case msg of+ Protocol.DaemonPong -> do+ writeIORef lastPongRef =<< getCurrentTime+ loop+ Protocol.DaemonBye -> exitSuccess where- waitForResponse sock = do- -- Wait for the socket to close- bs <- Socket.BS.recv sock 4096 `catchAny` (\_ -> return BS.empty)+ runPingThread lastPongRef rx tx = go+ where+ go = do+ timestamp <- getCurrentTime+ lastPong <- readIORef lastPongRef - -- A zero-length response means that the daemon has closed the socket- guard $ not $ BS.null bs+ if timestamp >= addUTCTime 20 lastPong+ then atomically $ writeTBMQueue rx (Left SocketStalled)+ else do+ atomically $ writeTBMQueue tx Protocol.ClientPing+ threadDelay (2 * 1000 * 1000)+ go - case Aeson.eitherDecodeStrict bs of- Left err -> putErrText (toS err)- Right DaemonBye -> return ()+ handleOutgoing tx sock = go+ where+ go = do+ mmsg <- atomically $ readTBMQueue tx+ case mmsg of+ Nothing -> return ()+ Just msg -> do+ Retry.retryAll $ const $ Socket.LBS.sendAll sock $ Aeson.encode msg `Lazy.Char8.snoc` '\n'+ go + handleIncoming rx sock = go+ where+ go = do+ -- Wait for the socket to close+ bs <- Socket.BS.recv sock 4096 `catchAny` (\_ -> return BS.empty)++ -- A zero-length response means that the daemon has closed the socket+ if BS.null bs+ then atomically $ writeTBMQueue rx (Left SocketClosed)+ else case Aeson.eitherDecodeStrict bs of+ Left err -> do+ let terr = toS err+ putErrText terr+ atomically $ writeTBMQueue rx (Left (SocketDecodingError terr))+ Right msg -> do+ atomically $ writeTBMQueue rx (Right msg)+ go+ withDaemonConn :: Maybe FilePath -> (Socket.Socket -> IO a) -> IO a withDaemonConn optionalSocketPath f = do socketPath <- maybe getSocketPath pure optionalSocketPath- bracket (open socketPath) Socket.close f `onException` failedToConnectTo socketPath+ bracket (open socketPath `onException` failedToConnectTo socketPath) Socket.close f where open socketPath = do sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol- Socket.connect sock (Socket.SockAddrUnix socketPath)-- -- Network.Socket.accept sets the socket to non-blocking by default.- Socket.withFdSocket sock $ \fd ->- Posix.setFdOption (fromIntegral fd) Posix.NonBlockingRead False-+ Retry.retryAll $ const $ Socket.connect sock (Socket.SockAddrUnix socketPath) return sock failedToConnectTo :: FilePath -> IO () failedToConnectTo socketPath = do- putErrText "Failed to connect to Cachix Daemon"+ putErrText "\nFailed to connect to Cachix Daemon" putErrText $ "Tried to connect to: " <> toS socketPath <> "\n"
+ src/Cachix/Client/Daemon/EventLoop.hs view
@@ -0,0 +1,73 @@+module Cachix.Client.Daemon.EventLoop+ ( new,+ send,+ sendIO,+ run,+ exitLoopWith,+ EventLoop,+ DaemonEvent (..),+ )+where++import Cachix.Client.Daemon.Types.EventLoop (DaemonEvent (..), EventLoop (..))+import Control.Concurrent.STM.TBMQueue+ ( isFullTBMQueue,+ newTBMQueueIO,+ readTBMQueue,+ tryWriteTBMQueue,+ )+import Data.Text.Lazy.Builder (toLazyText)+import qualified Katip+import Protolude++new :: (MonadIO m) => m EventLoop+new = do+ exitLatch <- liftIO newEmptyMVar+ queue <- liftIO $ newTBMQueueIO 100+ return $ EventLoop {queue, exitLatch}++-- | Send an event to the event loop with logging.+send :: (Katip.KatipContext m) => EventLoop -> DaemonEvent -> m ()+send = send' Katip.logFM++-- | Same as 'send', but does not require a 'Katip.KatipContext'.+sendIO :: forall m. (MonadIO m) => EventLoop -> DaemonEvent -> m ()+sendIO = send' logger+ where+ logger :: Katip.Severity -> Katip.LogStr -> m ()+ logger Katip.ErrorS msg = liftIO $ hPutStrLn stderr (toLazyText $ Katip.unLogStr msg)+ logger _ _ = return ()++send' :: (MonadIO m) => (Katip.Severity -> Katip.LogStr -> m ()) -> EventLoop -> DaemonEvent -> m ()+send' logger eventloop@(EventLoop {queue}) event = do+ res <- liftIO $ atomically $ tryWriteTBMQueue queue event+ case res of+ -- The queue is closed.+ Nothing ->+ logger Katip.DebugS "Ignored an event because the event loop is closed"+ -- Successfully wrote to the queue+ Just True -> return ()+ -- Failed to write to the queue+ Just False -> do+ isFull <- liftIO $ atomically $ isFullTBMQueue queue+ let message =+ if isFull+ then "Event loop is full"+ else "Unknown error"+ logger Katip.ErrorS $ "Failed to write to event loop: " <> message+ exitLoopWith (ExitFailure 1) eventloop++run :: (MonadIO m) => EventLoop -> (DaemonEvent -> m ()) -> m ExitCode+run eventloop f = do+ fix $ \loop -> do+ mevent <- liftIO $ atomically $ readTBMQueue (queue eventloop)+ case mevent of+ Just event -> f event+ Nothing -> exitLoopWith (ExitFailure 1) eventloop++ liftIO (tryReadMVar (exitLatch eventloop)) >>= \case+ Just exitCode -> return exitCode+ Nothing -> loop++exitLoopWith :: (MonadIO m) => ExitCode -> EventLoop -> m ()+exitLoopWith exitCode EventLoop {exitLatch} = void $ liftIO $ tryPutMVar exitLatch exitCode
src/Cachix/Client/Daemon/Listen.hs view
@@ -1,5 +1,6 @@ module Cachix.Client.Daemon.Listen ( listen,+ handleClient, serverBye, getSocketPath, openSocket,@@ -8,11 +9,17 @@ where import Cachix.Client.Config.Orphans ()+import qualified Cachix.Client.Daemon.EventLoop as EventLoop import Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.Types.EventLoop (EventLoop)+import Cachix.Client.Daemon.Types.SocketStore (SocketId) import Control.Exception.Safe (catchAny)-import qualified Control.Exception.Safe as Safe+import qualified Control.Monad.Catch as E import qualified Data.Aeson as Aeson+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as Char8 import qualified Katip+import Network.Socket (Socket) import qualified Network.Socket as Socket import qualified Network.Socket.ByteString as Socket.BS import qualified Network.Socket.ByteString.Lazy as Socket.LBS@@ -25,8 +32,9 @@ ) import qualified System.Environment as System import System.FilePath ((</>))-import System.IO.Error (isDoesNotExistError)+import System.IO.Error (isDoesNotExistError, isResourceVanishedError) +-- TODO: reconcile with Client data ListenError = SocketError SomeException | DecodingError Text@@ -35,59 +43,68 @@ instance Exception ListenError where displayException = \case SocketError err -> "Failed to read from the daemon socket: " <> show err- DecodingError err -> "Failed to decode request: " <> toS err+ DecodingError err -> "Failed to decode request:\n" <> toS err --- | The main daemon server loop.+-- | Listen for incoming connections on the given socket path. 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+ (E.MonadMask m, Katip.KatipContext m) =>+ EventLoop ->+ FilePath ->+ m ()+listen eventloop daemonSocketPath = forever $ do+ E.bracketOnError (openSocket daemonSocketPath) closeSocket $ \sock -> do+ liftIO $ Socket.listen sock Socket.maxListenQueue+ (conn, _peerAddr) <- liftIO $ Socket.accept sock+ EventLoop.send eventloop (EventLoop.AddSocketClient conn)++-- | Handle incoming messages from a client.+--+-- Automatically responds to pings.+-- Requests the daemon to remove the client socket once the loop exits.+handleClient ::+ forall m.+ (E.MonadMask m, Katip.KatipContext m) =>+ EventLoop ->+ SocketId ->+ Socket ->+ m ()+handleClient eventloop socketId conn = do+ go `E.finally` removeClient where- loop = do- result <- liftIO $ runExceptT $ readPushRequest sock+ go = do+ ebs <- liftIO $ try $ Socket.BS.recv conn 4096 - case result of- Right (ClientStop, clientConn) -> do- onClientStop- return clientConn- Right (ClientPushRequest pushRequest, clientConn) -> do- onPushRequest pushRequest clientConn- loop- Left err@(DecodingError _) -> do- Katip.logFM Katip.ErrorS $ Katip.ls $ displayException err- loop- Left err -> throwIO err+ case ebs of+ Left err | isResourceVanishedError err -> return ()+ Left _ -> return ()+ -- If the socket returns 0 bytes, then it is closed+ Right bs | BS.null bs -> return ()+ Right bs -> do+ msgs <- catMaybes <$> mapM decodeMessage (Char8.split '\n' bs) -serverBye :: Socket.Socket -> IO ()-serverBye sock =- Socket.LBS.sendAll sock (Aeson.encode DaemonBye) `catchAny` (\_ -> return ())+ forM_ msgs $ \msg -> do+ EventLoop.send eventloop (EventLoop.ReceivedMessage msg)+ case msg of+ Protocol.ClientPing ->+ liftIO $ Socket.LBS.sendAll conn (Aeson.encode DaemonPong)+ _ -> return () --- | Try to read and decode a push request.-readPushRequest :: Socket.Socket -> ExceptT ListenError IO (ClientMessage, Socket.Socket)-readPushRequest sock = do- (bs, clientConn) <- readFromSocket `mapSyncException` SocketError- decodeMessage clientConn bs- where- readFromSocket = liftIO $ do- -- NOTE: this sets up a non-blocking socket to the client- (conn, _peerAddr) <- Socket.accept sock- bs <- Socket.BS.recv conn 4096- return (bs, conn)+ go - decodeMessage conn bs =+ decodeMessage :: ByteString -> m (Maybe Protocol.ClientMessage)+ decodeMessage "" = return Nothing+ decodeMessage bs = case Aeson.eitherDecodeStrict bs of- Left err -> throwE $ DecodingError (toS err)- Right pushRequest -> return (pushRequest, conn)+ Left err -> do+ Katip.logFM Katip.ErrorS $ Katip.ls $ displayException (DecodingError (toS err))+ return Nothing+ Right msg -> return (Just msg) -mapSyncException :: (Exception e1, Exception e2, Safe.MonadCatch m) => m a -> (e1 -> e2) -> m a-mapSyncException a f = a `Safe.catch` (Safe.throwM . f)+ removeClient = EventLoop.send eventloop (EventLoop.RemoveSocketClient socketId)++serverBye :: Socket.Socket -> IO ()+serverBye sock =+ Socket.LBS.sendAll sock (Aeson.encode DaemonBye) `catchAny` (\_ -> return ()) getSocketPath :: IO FilePath getSocketPath = do
src/Cachix/Client/Daemon/PostBuildHook.hs view
@@ -1,8 +1,22 @@ {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE TypeApplications #-} -module Cachix.Client.Daemon.PostBuildHook where+module Cachix.Client.Daemon.PostBuildHook+ ( -- * Post-build hook setup+ PostBuildHookEnv (..),+ withSetup, + -- * Set up env vars+ EnvVar,+ modifyEnv,++ -- * Internal+ buildNixConfEnv,+ buildNixUserConfFilesEnv,+ )+where++import Control.Monad.Catch (MonadMask) import Data.Containers.ListUtils (nubOrd) import Data.String (String) import Data.String.Here@@ -15,7 +29,7 @@ ) import System.Environment (getExecutablePath, lookupEnv) import System.FilePath ((</>))-import System.IO.Temp (withSystemTempDirectory)+import System.IO.Temp (getCanonicalTemporaryDirectory, withTempDirectory) import System.Posix.Files type EnvVar = (String, String)@@ -24,9 +38,17 @@ modifyEnv (envName, envValue) processEnv = nubOrd $ (envName, envValue) : processEnv -withSetup :: Maybe FilePath -> (FilePath -> EnvVar -> IO a) -> IO a+data PostBuildHookEnv = PostBuildHookEnv+ { tempDir :: !FilePath,+ postBuildHookConfigPath :: !FilePath,+ postBuildHookScriptPath :: !FilePath,+ daemonSock :: !FilePath,+ envVar :: !EnvVar+ }++withSetup :: Maybe FilePath -> (PostBuildHookEnv -> IO a) -> IO a withSetup mdaemonSock f =- withSystemTempDirectory "cachix-daemon" $ \tempDir -> do+ withRunnerFriendlyTempDirectory "cachix-daemon" $ \tempDir -> do let postBuildHookScriptPath = tempDir </> "post-build-hook.sh" postBuildHookConfigPath = tempDir </> "nix.conf" daemonSock = fromMaybe (tempDir </> "daemon.sock") mdaemonSock@@ -37,13 +59,13 @@ mnixConfEnv <- buildNixConfEnv postBuildHookScriptPath nixUserConfFilesEnv <- buildNixUserConfFilesEnv postBuildHookConfigPath- nixConfEnvVar <- case mnixConfEnv of+ envVar <- case mnixConfEnv of Just nixConfEnv -> return nixConfEnv Nothing -> do writeFile postBuildHookConfigPath (postBuildHookConfig postBuildHookScriptPath) return nixUserConfFilesEnv - f daemonSock nixConfEnvVar+ f PostBuildHookEnv {..} -- | Build the NIX_CONF environment variable. --@@ -118,3 +140,15 @@ --socket ${toS socketPath :: Text} \\ $OUT_PATHS |]++-- | Run an action with a temporary directory.+--+-- Respects the RUNNER_TEMP environment variable if set.+-- This is important on self-hosted GitHub runners with locked down system temp directories.+-- The directory is deleted after use.+withRunnerFriendlyTempDirectory :: (MonadIO m, MonadMask m) => String -> (FilePath -> m a) -> m a+withRunnerFriendlyTempDirectory name action = do+ runnerTempDir <- liftIO $ lookupEnv "RUNNER_TEMP"+ systemTempDir <- liftIO getCanonicalTemporaryDirectory+ let tempDir = maybe systemTempDir toS runnerTempDir+ withTempDirectory tempDir name action
src/Cachix/Client/Daemon/Protocol.hs view
@@ -16,15 +16,17 @@ -- | JSON messages that the client can send to the daemon data ClientMessage- = ClientPushRequest PushRequest+ = ClientPushRequest !PushRequest | ClientStop- deriving stock (Generic)+ | ClientPing+ deriving stock (Generic, Show) deriving anyclass (Aeson.FromJSON, Aeson.ToJSON) -- | JSON messages that the daemon can send to the client data DaemonMessage- = DaemonBye- deriving stock (Generic)+ = DaemonPong+ | DaemonBye+ deriving stock (Generic, Show) deriving anyclass (Aeson.FromJSON, Aeson.ToJSON) newtype PushRequestId = PushRequestId UUID
src/Cachix/Client/Daemon/PushManager.hs view
@@ -12,6 +12,7 @@ lookupPushJob, withPushJob, resolvePushJob,+ pendingJobCount, -- * Store paths queueStorePaths,@@ -27,6 +28,9 @@ pushStorePathProgress, pushStorePathDone, pushStorePathFailed,++ -- * Helpers+ atomicallyWithTimeout, ) where @@ -44,6 +48,7 @@ import Cachix.Client.Retry (retryAll) import qualified Cachix.Types.BinaryCache as BinaryCache import qualified Conduit as C+import qualified Control.Concurrent.Async as Async import Control.Concurrent.STM.TBMQueue import Control.Concurrent.STM.TVar import qualified Control.Monad.Catch as E@@ -54,7 +59,7 @@ import Data.IORef import qualified Data.Set as Set import qualified Data.Text as T-import Data.Time (getCurrentTime)+import Data.Time (UTCTime, diffUTCTime, getCurrentTime, secondsToNominalDiffTime) import Hercules.CNix (StorePath) import Hercules.CNix.Store (Store, parseStorePath, storePathToPath) import qualified Katip@@ -66,23 +71,26 @@ import qualified UnliftIO.QSem as QSem newPushManagerEnv :: (MonadIO m) => PushOptions -> Logger -> OnPushEvent -> m PushManagerEnv-newPushManagerEnv pushOptions pmLogger pmOnPushEvent = liftIO $ do+newPushManagerEnv pushOptions pmLogger onPushEvent = liftIO $ do pmPushJobs <- newTVarIO mempty+ pmPendingJobCount <- newTVarIO 0 pmStorePathReferences <- newTVarIO mempty pmTaskQueue <- newTBMQueueIO 1000 pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)+ pmLastEventTimestamp <- newTVarIO =<< getCurrentTime+ let pmOnPushEvent id pushEvent = updateTimestampTVar pmLastEventTimestamp >> onPushEvent id pushEvent+ return $ PushManagerEnv {..} runPushManager :: (MonadIO m) => PushManagerEnv -> PushManager a -> m a runPushManager env f = liftIO $ unPushManager f `runReaderT` env -stopPushManager :: PushManagerEnv -> IO ()-stopPushManager PushManagerEnv {pmTaskQueue, pmPushJobs} =- atomically $ do- pushJobs <- readTVar pmPushJobs- if all PushJob.isProcessed (HashMap.elems pushJobs)- then closeTBMQueue pmTaskQueue- else retry+stopPushManager :: TimeoutOptions -> PushManagerEnv -> IO ()+stopPushManager timeoutOptions PushManagerEnv {..} = do+ atomicallyWithTimeout timeoutOptions pmLastEventTimestamp $ do+ pendingJobs <- readTVar pmPendingJobCount+ check (pendingJobs <= 0)+ atomically $ closeTBMQueue pmTaskQueue -- Manage push jobs @@ -96,6 +104,7 @@ liftIO $ atomically $ do modifyTVar' pmPushJobs $ HashMap.insert (pushId pushJob) pushJob+ incrementTVar pmPendingJobCount writeTBMQueue pmTaskQueue $ ResolveClosure (pushId pushJob) return (pushId pushJob)@@ -103,7 +112,12 @@ removePushJob :: Protocol.PushRequestId -> PushManager () removePushJob pushId = do PushManagerEnv {..} <- ask- liftIO $ atomically $ modifyTVar' pmPushJobs $ HashMap.delete pushId+ liftIO $ atomically $ do+ mpushJob <- HashMap.lookup pushId <$> readTVar pmPushJobs+ for_ mpushJob $ \pushJob -> do+ -- Decrement the job count if this job had not been processed yet+ unless (PushJob.isProcessed pushJob) (decrementTVar pmPendingJobCount)+ modifyTVar' pmPushJobs (HashMap.delete pushId) lookupPushJob :: Protocol.PushRequestId -> PushManager (Maybe PushJob) lookupPushJob pushId = do@@ -120,7 +134,11 @@ modifyPushJob :: Protocol.PushRequestId -> (PushJob -> PushJob) -> PushManager (Maybe PushJob) modifyPushJob pushId f = do pushJobs <- asks pmPushJobs- liftIO $ atomically $ stateTVar pushJobs $ \jobs -> do+ 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) @@ -132,9 +150,17 @@ failPushJob :: Protocol.PushRequestId -> PushManager () failPushJob pushId = do+ PushManagerEnv {..} <- ask timestamp <- liftIO getCurrentTime- void $ modifyPushJob pushId $ PushJob.fail timestamp+ liftIO $ atomically $ do+ _ <- modifyPushJobSTM pmPushJobs pushId $ PushJob.fail timestamp+ decrementTVar pmPendingJobCount +pendingJobCount :: PushManager Int+pendingJobCount = do+ pmPendingJobCount <- asks pmPendingJobCount+ liftIO $ readTVarIO pmPendingJobCount+ -- Manage store paths queueStorePaths :: Protocol.PushRequestId -> [FilePath] -> PushManager ()@@ -164,11 +190,14 @@ checkPushJobCompleted :: Protocol.PushRequestId -> PushManager () checkPushJobCompleted pushId = do+ PushManagerEnv {..} <- ask mpushJob <- lookupPushJob pushId for_ mpushJob $ \pushJob -> when (Set.null $ pushQueue pushJob) $ do timestamp <- liftIO getCurrentTime- _ <- modifyPushJob pushId $ PushJob.complete timestamp+ liftIO $ atomically $ do+ _ <- modifyPushJobSTM pmPushJobs pushId $ PushJob.complete timestamp+ decrementTVar pmPendingJobCount pushFinished pushJob queuedStorePathCount :: PushManager Integer@@ -401,9 +430,6 @@ -- Helpers -transactionally :: (Foldable t, MonadIO m) => t (STM ()) -> m ()-transactionally = liftIO . atomically . sequence_- storeToFilePath :: (MonadIO m) => Store -> StorePath -> m FilePath storeToFilePath store storePath = do fp <- liftIO $ storePathToPath store storePath@@ -418,3 +444,48 @@ withException :: (E.MonadCatch m) => m a -> (SomeException -> m a) -> m a withException action handler = action `E.catchAll` (\e -> handler e >> E.throwM e)++-- STM helpers++transactionally :: (Foldable t, MonadIO m) => t (STM ()) -> m ()+transactionally = liftIO . atomically . sequence_++updateTimestampTVar :: (MonadIO m) => TVar UTCTime -> m ()+updateTimestampTVar tvar = liftIO $ do+ now <- getCurrentTime+ atomically $ writeTVar tvar now++incrementTVar :: TVar Int -> STM ()+incrementTVar tvar = modifyTVar' tvar (+ 1)++decrementTVar :: TVar Int -> STM ()+decrementTVar tvar = modifyTVar' tvar (subtract 1)++-- | Run a transaction with a timeout.+atomicallyWithTimeout ::+ TimeoutOptions ->+ -- | A TVar timestamp to compare against+ TVar UTCTime ->+ -- | The transaction to run+ STM () ->+ IO ()+atomicallyWithTimeout TimeoutOptions {..} timeVar transaction = do+ timeoutVar <- newTVarIO False+ Async.race_+ (updateShutdownTimeout timeoutVar)+ (waitForGracefulShutdown timeoutVar)+ where+ waitForGracefulShutdown timeout =+ atomically $ transaction `orElse` checkShutdownTimeout timeout++ updateShutdownTimeout timeoutVar =+ forever $ do+ now <- getCurrentTime+ atomically $ do+ timestamp <- readTVar timeVar+ let isTimeout =+ secondsToNominalDiffTime (realToFrac toTimeout) <= now `diffUTCTime` timestamp+ writeTVar timeoutVar isTimeout+ threadDelay $ ceiling (toPollingInterval * 1000.0 * 1000.0)++ checkShutdownTimeout timeout = check =<< readTVar timeout
src/Cachix/Client/Daemon/PushManager/PushJob.hs view
@@ -58,16 +58,14 @@ markStorePathPushed :: FilePath -> PushJob -> PushJob markStorePathPushed storePath pushJob@(PushJob {pushQueue, pushResult}) = pushJob- { pushStatus = Running,- pushQueue = Set.delete storePath pushQueue,+ { pushQueue = Set.delete storePath pushQueue, pushResult = addPushedPath storePath pushResult } markStorePathFailed :: FilePath -> PushJob -> PushJob markStorePathFailed storePath pushJob@(PushJob {pushQueue, pushResult}) = pushJob- { pushStatus = Running,- pushQueue = Set.delete storePath pushQueue,+ { pushQueue = Set.delete storePath pushQueue, pushResult = addFailedPath storePath pushResult }
+ src/Cachix/Client/Daemon/SocketStore.hs view
@@ -0,0 +1,43 @@+module Cachix.Client.Daemon.SocketStore+ ( newSocketStore,+ addSocket,+ removeSocket,+ toList,+ Socket (..),+ )+where++import Cachix.Client.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..))+import Control.Concurrent.STM.TVar+import Control.Monad.IO.Unlift (MonadUnliftIO)+import qualified Data.HashMap.Strict as HashMap+import qualified Data.UUID.V4 as UUID+import qualified Network.Socket+import Protolude hiding (toList)+import qualified UnliftIO.Async as Async++newSocketStore :: (MonadIO m) => m SocketStore+newSocketStore = SocketStore <$> liftIO (newTVarIO mempty)++newSocketId :: (MonadIO m) => m SocketId+newSocketId = liftIO UUID.nextRandom++addSocket :: (MonadUnliftIO m) => Network.Socket.Socket -> (SocketId -> Network.Socket.Socket -> m ()) -> SocketStore -> m ()+addSocket socket handler (SocketStore st) = do+ socketId <- newSocketId+ handlerThread <- Async.async (handler socketId socket)+ liftIO $ atomically $ modifyTVar' st $ HashMap.insert socketId (Socket {..})++removeSocket :: (MonadIO m) => SocketId -> SocketStore -> m ()+removeSocket socketId (SocketStore stvar) = do+ msocket <- liftIO $ atomically $ stateTVar stvar $ \st ->+ let msocket = HashMap.lookup socketId st+ in (msocket, HashMap.delete socketId st)++ -- shut down the handler thread+ mapM_ (Async.uninterruptibleCancel . handlerThread) msocket++toList :: (MonadIO m) => SocketStore -> m [Socket]+toList (SocketStore st) = do+ hm <- liftIO $ readTVarIO st+ return $ HashMap.elems hm
src/Cachix/Client/Daemon/Types/Daemon.hs view
@@ -13,9 +13,11 @@ import qualified Cachix.Client.Daemon.Protocol as Protocol import Cachix.Client.Daemon.ShutdownLatch (ShutdownLatch) import Cachix.Client.Daemon.Subscription (SubscriptionManager)+import Cachix.Client.Daemon.Types.EventLoop (EventLoop) import Cachix.Client.Daemon.Types.Log (Logger) import Cachix.Client.Daemon.Types.PushEvent (PushEvent) import Cachix.Client.Daemon.Types.PushManager (PushManagerEnv (..))+import Cachix.Client.Daemon.Types.SocketStore (SocketStore) import Cachix.Client.Env as Env import Cachix.Client.OptionsParser (PushOptions) import Cachix.Client.Push@@ -29,10 +31,14 @@ data DaemonEnv = DaemonEnv { -- | Cachix client env daemonEnv :: Env,+ -- | The main event loop+ daemonEventLoop :: EventLoop, -- | Push options, like compression settings and number of jobs daemonPushOptions :: PushOptions, -- | Path to the socket that the daemon listens on daemonSocketPath :: FilePath,+ -- | Main inbound socket thread+ daemonSocketThread :: MVar (Async ()), -- | The push secret for the binary cache daemonPushSecret :: PushSecret, -- | The name of the binary cache to push to@@ -41,6 +47,8 @@ daemonBinaryCache :: BinaryCache, -- | The state of active push requests daemonPushManager :: PushManagerEnv,+ -- | Connected clients over the socket+ daemonClients :: SocketStore, -- | A multiplexer for push events. daemonSubscriptionManager :: SubscriptionManager Protocol.PushRequestId PushEvent, -- | Logging env
+ src/Cachix/Client/Daemon/Types/EventLoop.hs view
@@ -0,0 +1,35 @@+module Cachix.Client.Daemon.Types.EventLoop+ ( EventLoop (..),+ ExitLatch,+ DaemonEvent (..),+ )+where++import qualified Cachix.Client.Daemon.Protocol as Protocol+import Cachix.Client.Daemon.Types.SocketStore (SocketId)+import Control.Concurrent.STM.TBMQueue (TBMQueue)+import Network.Socket (Socket)+import Protolude++-- | An event loop that processes 'DaemonEvent's.+data EventLoop = EventLoop+ { queue :: TBMQueue DaemonEvent,+ exitLatch :: ExitLatch+ }++-- | An exit latch is a semaphore that signals the event loop to exit.+-- The exit code should be returned by the 'EventLoop'.+type ExitLatch = MVar ExitCode++-- | Daemon events that are handled by the 'EventLoop'.+data DaemonEvent+ = -- | Shut down the daemon gracefully.+ ShutdownGracefully+ | -- | Re-establish the daemon socket+ ReconnectSocket+ | -- | Add a new client socket connection.+ AddSocketClient Socket+ | -- | Remove an existing client socket connection. For example, after it is closed.+ RemoveSocketClient SocketId+ | -- | Handle a new message from a client.+ ReceivedMessage Protocol.ClientMessage
src/Cachix/Client/Daemon/Types/PushEvent.hs view
@@ -13,7 +13,8 @@ import Protolude data PushEvent = PushEvent- { eventTimestamp :: UTCTime,+ { -- TODO: newtype a monotonic clock+ eventTimestamp :: UTCTime, eventPushId :: Protocol.PushRequestId, eventMessage :: PushEventMessage }
src/Cachix/Client/Daemon/Types/PushManager.hs view
@@ -5,12 +5,14 @@ module Cachix.Client.Daemon.Types.PushManager ( PushManagerEnv (..), PushManager (..),+ PushJobStore, PushJob (..), JobStatus (..), JobStats (..), PushResult (..), OnPushEvent, Task (..),+ TimeoutOptions (..), ) where @@ -48,6 +50,10 @@ pmTaskSemaphore :: QSem, -- | A callback for push events. pmOnPushEvent :: OnPushEvent,+ -- | The timestamp of the most recent event. This is used to track activity internally.+ pmLastEventTimestamp :: TVar UTCTime,+ -- | The number of pending (uncompleted) jobs.+ pmPendingJobCount :: TVar Int, pmLogger :: Logger } @@ -115,5 +121,13 @@ { jsCreatedAt :: UTCTime, jsStartedAt :: Maybe UTCTime, jsCompletedAt :: Maybe UTCTime+ }+ deriving stock (Eq, Show)++data TimeoutOptions = TimeoutOptions+ { -- | The maximum time to wait in seconds.+ toTimeout :: Float,+ -- | The interval at which to check the timeout condition.+ toPollingInterval :: Float } deriving stock (Eq, Show)
+ src/Cachix/Client/Daemon/Types/SocketStore.hs view
@@ -0,0 +1,23 @@+module Cachix.Client.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..)) where++import Control.Concurrent.STM.TVar (TVar)+import Data.HashMap.Strict (HashMap)+import Data.UUID (UUID)+import qualified Network.Socket as Network (Socket)+import Protolude++data Socket = Socket+ { socketId :: SocketId,+ socket :: Network.Socket,+ handlerThread :: Async ()+ }++instance Eq Socket where+ (==) = (==) `on` socketId++instance Ord Socket where+ compare = comparing socketId++type SocketId = UUID++newtype SocketStore = SocketStore (TVar (HashMap SocketId Socket))
src/Cachix/Client/Daemon/Worker.hs view
@@ -3,6 +3,7 @@ stopWorkers, startWorker, stopWorker,+ Immortal.Thread, ) where
src/Cachix/Client/NixVersion.hs view
@@ -1,33 +1,64 @@+-- TODO: we may need to revisit this when the various flavours of Nix start to diverge module Cachix.Client.NixVersion ( assertNixVersion, parseNixVersion,+ minimalVersion, ) where +import Data.Either.Extra (mapLeft) import Data.Text as T-import Data.Versions+import Data.Versions hiding (versioning') import Protolude import System.Process (readProcessWithExitCode)+import Text.Megaparsec (Parsec, anySingle, choice, eof, lookAhead, manyTill, parse)+import qualified Text.Megaparsec as P+import Text.Megaparsec.Char (digitChar) +minimalVersion :: SemVer+minimalVersion =+ semver "2.0.1" & fromRight (panic "Couldn't parse minimalVersion")+ assertNixVersion :: IO (Either Text ()) assertNixVersion = do+ nixVersion <- fmap (>>= parseNixVersion) getRawNixVersion+ return $ case nixVersion of+ Left err -> Left err+ Right ver+ | ver < Ideal minimalVersion -> Left "Nix 2.0.2 or lower is not supported. Please upgrade: https://nixos.org/nix/"+ | otherwise -> Right ()++getRawNixVersion :: IO (Either Text Text)+getRawNixVersion = do (exitcode, out, err) <- readProcessWithExitCode "nix-env" ["--version"] mempty unless (err == "") $ putStrLn $ "nix-env stderr: " <> err return $ case exitcode of ExitFailure i -> Left $ "'nix-env --version' exited with " <> show i- ExitSuccess -> parseNixVersion $ toS out+ ExitSuccess -> Right (toS out) -parseNixVersion :: Text -> Either Text ()-parseNixVersion output =- let verStr = T.drop 14 $ T.strip output- err = "Couldn't parse 'nix-env --version' output: " <> output- in case versioning verStr of- Left _ -> Left err- Right ver- | verStr == "" -> Left err- | ver < Ideal minimalVersion -> Left "Nix 2.0.2 or lower is not supported. Please upgrade: https://nixos.org/nix/"- | otherwise -> Right ()+parseNixVersion :: Text -> Either Text Versioning+parseNixVersion input =+ mapLeft fromParsingError $ parse nixVersionParser "Nix version" input+ where+ fromParsingError pe =+ unlines+ [ "Couldn't parse 'nix-env --version' output: " <> input,+ T.pack $ errorBundlePretty pe+ ] -minimalVersion :: SemVer-minimalVersion =- semver "2.0.1" & fromRight (panic "Couldn't parse minimalVersion")+-- | Parses a semver string out of the output of `nix-env --version` or `nix --version`.+nixVersionParser :: Parsec Void Text Versioning+nixVersionParser = do+ _ <- manyTill anySingle (lookAhead digitChar)+ v <- versioning'+ _ <- manyTill anySingle eof+ pure v++-- | Same as `versioning'`, but without the restriction of preceding eof.+versioning' :: Parsec Void Text Versioning+versioning' =+ choice+ [ P.try (fmap Ideal semver'),+ P.try (fmap General version'),+ fmap Complex mess'+ ]
src/Cachix/Deploy/Lock.hs view
@@ -1,6 +1,7 @@ module Cachix.Deploy.Lock ( defaultLockDirectory, getLockDirectory,+ newLockFilePath, readPidFile, withTryLock, withTryLockAndPid,@@ -24,6 +25,11 @@ defaultLockDirectory :: FilePath defaultLockDirectory = "cachix" </> "deploy" </> "locks"++newLockFilePath :: FilePath -> IO FilePath+newLockFilePath path = do+ lockDirectory <- getLockDirectory+ return $ lockDirectory </> path <.> lockExtension getLockDirectory :: IO FilePath getLockDirectory = do
test/Daemon/PushManagerSpec.hs view
@@ -6,6 +6,8 @@ import qualified Cachix.Client.Daemon.PushManager.PushJob as PushJob import Cachix.Client.Daemon.Types.PushManager import Cachix.Client.OptionsParser (defaultPushOptions)+import Control.Concurrent.Async (concurrently_)+import Control.Concurrent.STM.TVar import Control.Retry (defaultRetryStatus) import qualified Data.Set as Set import Data.Time (getCurrentTime)@@ -113,6 +115,41 @@ prSkippedPaths = mempty } + describe "graceful shutdown" $ do+ it "shuts down with no jobs" $ do+ logger <- Log.new "daemon" Nothing Log.Debug+ pm <- newPushManagerEnv defaultPushOptions logger mempty+ stopPushManager timeoutOptions pm++ it "shuts down after jobs complete" $ do+ let paths = ["foo"]+ let longTimeoutOptions = TimeoutOptions {toTimeout = 10.0, toPollingInterval = 0.1}+ logger <- Log.new "daemon" Nothing Log.Debug+ pm <- newPushManagerEnv defaultPushOptions logger mempty++ _ <- runPushManager pm $ do+ let request = Protocol.PushRequest {Protocol.storePaths = paths}+ addPushJob request++ concurrently_ (stopPushManager longTimeoutOptions pm) $+ runPushManager pm $+ for_ paths pushStorePathDone++ it "shuts down on job stall" $ do+ logger <- Log.new "daemon" Nothing Log.Debug+ pm <- newPushManagerEnv defaultPushOptions logger mempty+ _ <- runPushManager pm $ do+ let request = Protocol.PushRequest {Protocol.storePaths = ["foo"]}+ addPushJob request++ stopPushManager timeoutOptions pm++ describe "STM" $+ describe "timeout" $ do+ it "times out a transaction after n seconds" $ do+ timestamp <- newTVarIO =<< getCurrentTime+ atomicallyWithTimeout timeoutOptions timestamp retry+ withPushManager :: (PushManagerEnv -> IO a) -> IO a withPushManager f = do logger <- liftIO $ Log.new "daemon" Nothing Log.Debug@@ -120,3 +157,6 @@ inPushManager :: PushManager a -> IO a inPushManager f = withPushManager (`runPushManager` f)++timeoutOptions :: TimeoutOptions+timeoutOptions = TimeoutOptions {toTimeout = 0.2, toPollingInterval = 0.1}
test/NixVersionSpec.hs view
@@ -1,23 +1,39 @@-module NixVersionSpec where+module NixVersionSpec (spec) where import Cachix.Client.NixVersion+import Data.Versions (Versioning, versioning) import Protolude import Test.Hspec spec :: Spec-spec = describe "parseNixVersion" $ do- it "parses 'nix-env (Nix) 2.0' as Nix20" $- parseNixVersion "nix-env (Nix) 2.0"- `shouldBe` Left "Nix 2.0.2 or lower is not supported. Please upgrade: https://nixos.org/nix/"- it "parses 'nix-env (Nix) 2.0.1' as Nix201" $- parseNixVersion "nix-env (Nix) 2.0.1"- `shouldBe` Right ()- it "parses 'nix-env (Nix) 1.11.13' as Nix1XX" $- parseNixVersion "nix-env (Nix) 1.11.13"- `shouldBe` Left "Nix 2.0.2 or lower is not supported. Please upgrade: https://nixos.org/nix/"- it "parses 'nix-env (Nix) 2.0.2' as Nix202" $- parseNixVersion "nix-env (Nix) 2.0.2"- `shouldBe` Right ()- it "fails with unknown string 'foobar'" $- parseNixVersion "foobar"- `shouldBe` Left "Couldn't parse 'nix-env --version' output: foobar"+spec = do+ describe "parseNixVersion" $ do+ it "fails with unknown string 'foobar'" $+ parseNixVersion "foobar" `shouldSatisfy` isLeft++ it "parses out a semver out of an unexpected string" $+ parseNixVersion "foobar 2.0.1 other stuff"+ `shouldBe` Right (toVersioning "2.0.1")++ describe "Nix" $+ runParserTests+ [ ("nix-env (Nix) 2.0", toVersioning "2.0"),+ ("nix-env (Nix) 2.0.1", toVersioning "2.0.1"),+ ("nix-env (Nix) 2.0.1\n", toVersioning "2.0.1")+ ]++ describe "Lix" $+ runParserTests+ [ ("nix-env (Lix, like Nix) 2.90-beta.0", toVersioning "2.90-beta.0"),+ ("nix-env (Lix, like Nix) 2.90-beta.1-lixpre20240506-b6799ab", toVersioning "2.90-beta.1-lixpre20240506-b6799ab")+ ]++runParserTests :: [(Text, Versioning)] -> Spec+runParserTests tests = do+ forM_ tests $ \(input, expected) ->+ it ("parses '" <> toS input <> "'") $+ parseNixVersion input `shouldBe` Right expected++toVersioning :: Text -> Versioning+toVersioning str =+ versioning str & fromRight (panic "Couldn't parse version")