packages feed

cachix 1.9.1 → 1.9.2

raw patch · 10 files changed

+169/−62 lines, 10 filesdep ~hercules-ci-cnix-store

Dependency ranges changed: hercules-ci-cnix-store

Files

CHANGELOG.md view
@@ -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
cachix.cabal view
@@ -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
src/Cachix/Client.hs view
@@ -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
src/Cachix/Daemon.hs view
@@ -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 ()
src/Cachix/Daemon/Client.hs view
@@ -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)
src/Cachix/Daemon/PushManager.hs view
@@ -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
src/Cachix/Daemon/SocketStore.hs view
@@ -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
src/Cachix/Daemon/Subscription.hs view
@@ -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
src/Cachix/Daemon/Types/SocketStore.hs view
@@ -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
test/Daemon/PushManagerSpec.hs view
@@ -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