diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ 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.9.2] - 2025-12-11 + +### Changes + +- docs: document the daemon protocol +- dev: support building with cabal instead of stack +- deps: update constraints to require cnix-store >=0.4 + +### Fixed + +- daemon: address possible hangs when using `--wait` with `cachix daemon push` + ## [1.9.1] - 2025-10-20 ### Changed 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.9.1 +version: 1.9.2 synopsis: Command-line client for Nix binary cache hosting https://cachix.org @@ -153,7 +153,7 @@ , filepath , fsnotify >=0.4.1 , generic-lens - , hercules-ci-cnix-store + , hercules-ci-cnix-store >=0.4 , here , hnix-store-core >=0.8 , hnix-store-nar @@ -206,7 +206,7 @@ , wuss -- These versions are pinned in hercules-ci-cnix-store - pkgconfig-depends: nix-main, nix-store + pkgconfig-depends: nix-store executable cachix import: defaults diff --git a/src/Cachix/Client.hs b/src/Cachix/Client.hs --- a/src/Cachix/Client.hs +++ b/src/Cachix/Client.hs @@ -21,7 +21,6 @@ import Cachix.Deploy.Agent qualified as AgentCommand import Cachix.Deploy.OptionsParser qualified as DeployOptions import Hercules.CNix qualified as CNix -import Hercules.CNix.Util qualified as CNix.Util import Protolude import System.Console.AsciiProgress (displayConsoleRegions) import System.Posix.Signals qualified as Signal @@ -78,6 +77,3 @@ -- darwin: restore the signal mask modified by Nix -- https://github.com/cachix/cachix/issues/501 Signal.setSignalMask signalset - - -- Interrupt Nix before throwing UserInterrupt - CNix.Util.installDefaultSigINTHandler diff --git a/src/Cachix/Daemon.hs b/src/Cachix/Daemon.hs --- a/src/Cachix/Daemon.hs +++ b/src/Cachix/Daemon.hs @@ -89,7 +89,11 @@ daemonSubscriptionManagerThread <- newEmptyMVar daemonSubscriptionManager <- Subscription.newSubscriptionManager - let onPushEvent = Subscription.pushEvent daemonSubscriptionManager + let onPushEvent pushId event = do + Subscription.pushEvent daemonSubscriptionManager pushId event + case Types.eventMessage event of + Types.PushFinished -> Subscription.queueUnsubscribe daemonSubscriptionManager pushId + _ -> return () let pushParams = Push.newPushParams nixStore (clientenv daemonEnv) daemonBinaryCache daemonPushSecret daemonPushOptions daemonPushManager <- PushManager.newPushManagerEnv daemonPushOptions (daemonNarinfoQueryOptions daemonOptions) pushParams onPushEvent daemonLogger @@ -144,12 +148,20 @@ ReceivedMessage socketId clientMsg -> case clientMsg of ClientPushRequest pushRequest -> do - maybePushId <- queueJob pushRequest - case maybePushId of - Just pushId -> when (Protocol.subscribeToUpdates pushRequest) $ do + -- Create push job upfront so we can subscribe before queuing + pushJob <- liftIO $ PushManager.newPushJob pushRequest + let pushId = PushManager.pushId pushJob + + Katip.katipAddContext (Katip.sl "push_id" pushId) $ do + when (Protocol.subscribeToUpdates pushRequest) $ do chan <- liftIO $ subscribe daemon (Just pushId) - void $ liftIO $ Async.async $ publishToClient socketId chan daemon - Nothing -> Katip.logFM Katip.ErrorS "Failed to queue push request" + publisherThread <- liftIO $ Async.async $ publishToClient socketId chan daemon + SocketStore.addPublisherThread socketId pushId publisherThread daemonClients + + -- Now add the job + success <- queueJob pushJob + unless success $ + Katip.logFM Katip.ErrorS "Failed to queue push request" ClientStop -> do DaemonEnv {daemonOptions = opts, daemonClients = clients} <- ask if Options.daemonAllowRemoteStop opts @@ -225,10 +237,10 @@ threadDelay (15 * 1000 * 1000) -- 15 seconds throwTo mainThreadId ExitSuccess -queueJob :: Protocol.PushRequest -> Daemon (Maybe PushRequestId) -queueJob pushRequest = do +queueJob :: PushManager.PushJob -> Daemon Bool +queueJob pushJob = do DaemonEnv {daemonPushManager} <- ask - liftIO $ PushManager.runPushManager daemonPushManager (PushManager.addPushJob pushRequest) + liftIO $ PushManager.runPushManager daemonPushManager (PushManager.addPushJob pushJob) subscribe :: DaemonEnv -> Maybe PushRequestId -> IO (TMChan PushEvent) subscribe daemonEnv maybePushId = do @@ -243,14 +255,16 @@ -- Publish messages from the channel to the client over the socket publishToClient :: SocketId -> TMChan PushEvent -> DaemonEnv -> IO () -publishToClient socketId chan daemonEnv = do - forever $ do - msg <- atomically $ readTMChan chan - case msg of - Nothing -> return () - Just evt -> do - let daemonMsg = DaemonPushEvent evt - SocketStore.sendAll socketId (Protocol.newMessage daemonMsg) (daemonClients daemonEnv) +publishToClient socketId chan daemonEnv = go + where + go = do + msg <- atomically $ readTMChan chan + case msg of + Nothing -> return () + Just evt -> do + let daemonMsg = DaemonPushEvent evt + SocketStore.sendAll socketId (Protocol.newMessage daemonMsg) (daemonClients daemonEnv) + go -- | Print the daemon configuration to the log. printConfiguration :: Daemon () diff --git a/src/Cachix/Daemon/Client.hs b/src/Cachix/Daemon/Client.hs --- a/src/Cachix/Daemon/Client.hs +++ b/src/Cachix/Daemon/Client.hs @@ -159,7 +159,8 @@ then exitWith (toExitCode DaemonPushFailure) else exitSuccess _ -> loop - _ -> loop + Protocol.DaemonError err -> handleDaemonError err + Protocol.DaemonExit exitStatus -> handleDaemonExit exitStatus -- | Tell the daemon to stop and wait for it to gracefully exit stop :: Env -> DaemonOptions -> IO () @@ -178,16 +179,9 @@ Just (Right msg) -> case msg of Protocol.DaemonPong -> loop - Protocol.DaemonExit exitStatus -> - case exitCode exitStatus of - 0 -> exitSuccess - code -> exitWith (ExitFailure code) - Protocol.DaemonError errorMsg -> do - case errorMsg of - Protocol.UnsupportedCommand errMsg -> do - putErrText $ toS errMsg - exitWith (ExitFailure 2) - _ -> loop + Protocol.DaemonError err -> handleDaemonError err + Protocol.DaemonExit exitStatus -> handleDaemonExit exitStatus + Protocol.DaemonPushEvent _ -> loop withDaemonConn :: Maybe FilePath -> (Socket.Socket -> IO a) -> IO a withDaemonConn optionalSocketPath f = do @@ -204,3 +198,20 @@ failedToConnectTo socketPath = do putErrText "\nFailed to connect to Cachix Daemon" putErrText $ "Tried to connect to: " <> toS socketPath <> "\n" + +handleDaemonExit :: Protocol.DaemonExitStatus -> IO a +handleDaemonExit exitStatus = + case exitCode exitStatus of + 0 -> exitSuccess + code -> do + case exitMessage exitStatus of + Just msg -> putErrText $ toS msg + Nothing -> return () + exitWith (ExitFailure code) + +handleDaemonError :: Protocol.DaemonErrorMessage -> IO a +handleDaemonError err = + case err of + Protocol.UnsupportedCommand errMsg -> do + putErrText $ toS errMsg + exitWith (ExitFailure 2) 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 @@ -8,7 +8,9 @@ -- * Push job PushJob (..), + newPushJob, addPushJob, + addPushJobFromRequest, lookupPushJob, withPushJob, resolvePushJob, @@ -128,28 +130,36 @@ -- Manage push jobs -addPushJob :: Protocol.PushRequest -> PushManager (Maybe Protocol.PushRequestId) -addPushJob pushRequest = do +newPushJob :: (MonadIO m) => Protocol.PushRequest -> m PushJob +newPushJob = PushJob.new + +addPushJob :: PushJob -> PushManager Bool +addPushJob pushJob = do PushManagerEnv {..} <- ask - pushJob <- PushJob.new pushRequest + let pushId = PushJob.pushId pushJob - Katip.logLocM Katip.DebugS $ Katip.ls $ "Queued push job " <> (show (pushId pushJob) :: Text) + Katip.logLocM Katip.DebugS $ Katip.ls $ "Queued push job " <> (show pushId :: Text) let queueJob = do - modifyTVar' pmPushJobs $ HashMap.insert (pushId pushJob) pushJob + modifyTVar' pmPushJobs $ HashMap.insert pushId pushJob incrementTVar pmPendingJobCount - res <- tryWriteTask pmTaskQueue $ QueryMissingPaths (pushId pushJob) + res <- tryWriteTask pmTaskQueue $ QueryMissingPaths pushId case res of Just True -> return True _ -> return False didQueue <- liftIO $ atomically queueJob - if didQueue - then return $ Just $ pushId pushJob - else do - Katip.logLocM Katip.WarningS "Failed to queue push job. Queue likely full." - return Nothing + unless didQueue $ + Katip.logLocM Katip.WarningS "Failed to queue push job. Queue likely full." + + return didQueue + +addPushJobFromRequest :: Protocol.PushRequest -> PushManager (Maybe Protocol.PushRequestId) +addPushJobFromRequest pushRequest = do + pushJob <- newPushJob pushRequest + success <- addPushJob pushJob + return $ if success then Just (PushJob.pushId pushJob) else Nothing removePushJob :: Protocol.PushRequestId -> PushManager () removePushJob pushId = do diff --git a/src/Cachix/Daemon/SocketStore.hs b/src/Cachix/Daemon/SocketStore.hs --- a/src/Cachix/Daemon/SocketStore.hs +++ b/src/Cachix/Daemon/SocketStore.hs @@ -2,12 +2,15 @@ ( newSocketStore, addSocket, removeSocket, + addPublisherThread, + cancelPublisherThread, toList, Socket (..), sendAll, ) where +import Cachix.Daemon.Types.PushEvent (PushRequestId) import Cachix.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..)) import Control.Concurrent.STM.TVar import Control.Monad.IO.Unlift (MonadUnliftIO) @@ -29,6 +32,7 @@ addSocket socket handler (SocketStore st) = do socketId <- newSocketId handlerThread <- Async.async (handler socketId socket) + publisherThreads <- liftIO $ newTVarIO HashMap.empty liftIO $ atomically $ modifyTVar' st $ HashMap.insert socketId (Socket {..}) removeSocket :: (MonadIO m) => SocketId -> SocketStore -> m () @@ -37,8 +41,28 @@ let msocket = HashMap.lookup socketId st in (msocket, HashMap.delete socketId st) - -- shut down the handler thread - mapM_ (Async.uninterruptibleCancel . handlerThread) msocket + for_ msocket $ \sock -> do + -- Cancel all publisher threads for this socket + publishers <- liftIO $ readTVarIO (publisherThreads sock) + mapM_ Async.cancel (HashMap.elems publishers) + -- Cancel handler thread + Async.uninterruptibleCancel (handlerThread sock) + +addPublisherThread :: (MonadIO m) => SocketId -> PushRequestId -> Async () -> SocketStore -> m () +addPublisherThread socketId pushId thread (SocketStore stvar) = liftIO $ atomically $ do + sockets <- readTVar stvar + for_ (HashMap.lookup socketId sockets) $ \sock -> + modifyTVar' (publisherThreads sock) $ HashMap.insert pushId thread + +cancelPublisherThread :: (MonadIO m) => SocketId -> PushRequestId -> SocketStore -> m () +cancelPublisherThread socketId pushId (SocketStore stvar) = do + mthread <- liftIO $ atomically $ do + sockets <- readTVar stvar + case HashMap.lookup socketId sockets of + Nothing -> return Nothing + Just sock -> stateTVar (publisherThreads sock) $ \threads -> + (HashMap.lookup pushId threads, HashMap.delete pushId threads) + mapM_ Async.cancel mthread toList :: (MonadIO m) => SocketStore -> m [Socket] toList (SocketStore st) = do diff --git a/src/Cachix/Daemon/Subscription.hs b/src/Cachix/Daemon/Subscription.hs --- a/src/Cachix/Daemon/Subscription.hs +++ b/src/Cachix/Daemon/Subscription.hs @@ -1,4 +1,20 @@ -module Cachix.Daemon.Subscription where +module Cachix.Daemon.Subscription + ( SubscriptionManager (..), + Subscription (..), + newSubscriptionManager, + subscribeTo, + subscribeToAll, + subscribeToSTM, + subscribeToAllSTM, + queueUnsubscribe, + getSubscriptionsFor, + getSubscriptionsForSTM, + pushEvent, + pushEventSTM, + runSubscriptionManager, + stopSubscriptionManager, + ) +where import Control.Concurrent.STM.TBMQueue import Control.Concurrent.STM.TMChan @@ -13,9 +29,15 @@ data SubscriptionManager k v = SubscriptionManager { managerSubscriptions :: TVar (HashMap k (Seq.Seq (Subscription v))), managerGlobalSubscriptions :: TVar (Seq.Seq (Subscription v)), - managerEvents :: TBMQueue (k, v) + managerMessages :: TBMQueue (ManagerMessage k v) } +data ManagerMessage k v + = -- | An event to be dispatched to subscribers + EventMessage k v + | -- | Close subscriptions for a key (processed in order with events) + UnsubscribeMessage k + data Subscription v = -- | A subscriber that listens on a socket. SubSocket (TBMQueue v) Socket.Socket @@ -26,8 +48,8 @@ newSubscriptionManager = do subscriptions <- newTVarIO HashMap.empty globalSubscriptions <- newTVarIO Seq.empty - events <- newTBMQueueIO 10000 - pure $ SubscriptionManager subscriptions globalSubscriptions events + messages <- newTBMQueueIO 10000 + pure $ SubscriptionManager subscriptions globalSubscriptions messages -- Subscriptions @@ -60,6 +82,12 @@ subscriptions <- readTVar $ managerSubscriptions manager pure $ HashMap.lookupDefault Seq.empty key subscriptions +-- | Queue an unsubscribe command to close all subscription channels for a key. +-- This is processed in order with events, ensuring pending events are delivered first. +queueUnsubscribe :: (MonadIO m) => SubscriptionManager k v -> k -> m () +queueUnsubscribe manager key = + liftIO $ atomically $ writeTBMQueue (managerMessages manager) (UnsubscribeMessage key) + -- Events pushEvent :: (MonadIO m) => SubscriptionManager k v -> k -> v -> m () @@ -68,7 +96,7 @@ pushEventSTM :: SubscriptionManager k v -> k -> v -> STM () pushEventSTM manager key event = - writeTBMQueue (managerEvents manager) (key, event) + writeTBMQueue (managerMessages manager) (EventMessage key event) sendEventToSub :: Subscription v -> v -> STM () -- TODO: implement socket subscriptions. @@ -78,21 +106,31 @@ runSubscriptionManager :: (Show k, Show v, Hashable k, ToJSON v, MonadIO m) => SubscriptionManager k v -> m () runSubscriptionManager manager = do isDone <- liftIO $ atomically $ do - mevent <- readTBMQueue (managerEvents manager) - case mevent of + mmsg <- readTBMQueue (managerMessages manager) + case mmsg of Nothing -> return True - Just (key, event) -> do + Just (EventMessage key event) -> do subscriptions <- getSubscriptionsForSTM manager key globalSubscriptions <- readTVar $ managerGlobalSubscriptions manager mapM_ (`sendEventToSub` event) (subscriptions <> globalSubscriptions) return False + Just (UnsubscribeMessage key) -> do + subscriptions <- readTVar $ managerSubscriptions manager + case HashMap.lookup key subscriptions of + Nothing -> return () + Just subs -> do + forM_ subs $ \case + SubSocket queue _ -> closeTBMQueue queue + SubChannel chan -> closeTMChan chan + writeTVar (managerSubscriptions manager) $ HashMap.delete key subscriptions + return False unless isDone $ runSubscriptionManager manager stopSubscriptionManager :: SubscriptionManager k v -> IO () stopSubscriptionManager manager = do - liftIO $ atomically $ closeTBMQueue (managerEvents manager) + liftIO $ atomically $ closeTBMQueue (managerMessages manager) globalSubscriptions <- liftIO $ readTVarIO $ managerGlobalSubscriptions manager subscriptions <- liftIO $ readTVarIO $ managerSubscriptions manager diff --git a/src/Cachix/Daemon/Types/SocketStore.hs b/src/Cachix/Daemon/Types/SocketStore.hs --- a/src/Cachix/Daemon/Types/SocketStore.hs +++ b/src/Cachix/Daemon/Types/SocketStore.hs @@ -1,5 +1,6 @@ module Cachix.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..)) where +import Cachix.Daemon.Types.PushEvent (PushRequestId) import Control.Concurrent.STM.TVar (TVar) import Data.HashMap.Strict (HashMap) import Data.UUID (UUID) @@ -9,7 +10,8 @@ data Socket = Socket { socketId :: SocketId, socket :: Network.Socket, - handlerThread :: Async () + handlerThread :: Async (), + publisherThreads :: TVar (HashMap PushRequestId (Async ())) } instance Eq Socket where diff --git a/test/Daemon/PushManagerSpec.hs b/test/Daemon/PushManagerSpec.hs --- a/test/Daemon/PushManagerSpec.hs +++ b/test/Daemon/PushManagerSpec.hs @@ -113,7 +113,7 @@ describe "push manager" $ do it "queues push jobs " $ inPushManager $ do let request = Protocol.PushRequest {Protocol.storePaths = ["foo", "bar"], Protocol.subscribeToUpdates = False} - Just pushId <- addPushJob request + Just pushId <- addPushJobFromRequest request Just pushJob <- lookupPushJob pushId liftIO $ do PushJob.pushId pushJob `shouldBe` pushId @@ -123,7 +123,7 @@ let paths = ["bar", "foo"] let pushRequest = Protocol.PushRequest {Protocol.storePaths = paths, Protocol.subscribeToUpdates = False} - Just pushId <- addPushJob pushRequest + Just pushId <- addPushJobFromRequest pushRequest let pathSet = Set.fromList paths closure = PushJob.ResolvedClosure pathSet pathSet @@ -158,7 +158,7 @@ Just _ <- runPushManager pm $ do let request = Protocol.PushRequest {Protocol.storePaths = paths, Protocol.subscribeToUpdates = False} - pushId <- addPushJob request + pushId <- addPushJobFromRequest request let pathSet = Set.fromList paths closure = PushJob.ResolvedClosure pathSet pathSet for_ pushId $ \pid -> resolvePushJob pid closure @@ -177,7 +177,7 @@ withPushManager $ \pm -> do _ <- runPushManager pm $ do let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = False} - addPushJob request + addPushJobFromRequest request stopPushManager timeoutOptions pm