packages feed

cachix 1.10.0 → 1.11.0

raw patch · 27 files changed

+527/−226 lines, 27 files

Files

CHANGELOG.md view
@@ -5,6 +5,36 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [1.11.0] - 2026-03-20++### Added++- daemon: add `--keep-alive-interval` and `--keep-alive-timeout` CLI options (set interval to 0 to disable keep-alive)+- daemon: emit `PushStorePathSkipped` events for paths already present in the cache+- daemon: emit `PushStorePathInvalid` events for paths that fail validation+- daemon: detect and warn when graceful shutdown fails++### Changed++- daemon: serialize socket writes per client to prevent interleaved messages under load+- daemon: use a priority queue for pong, exit, and error messages so they are never blocked by a full progress queue+- daemon: drop progress events when client receive queue is full instead of blocking+- daemon: throttle progress events using monotonic time instead of byte thresholds+- daemon: keep-alive defaults changed from 2s/20s to 30s/180s++### Fixed++- daemon: fall back to system temp directory when `RUNNER_TEMP` would exceed the Unix socket path length limit+- daemon: fix shutdown order to ensure queued paths are not dropped+- daemon: fix immediate exit on second SIGINT+- daemon: fix progress event reporting incorrect delta bytes++## [1.10.1] - 2026-01-13++### Fixed++- `watch-exec`: respect `CACHIX_DAEMON_SOCKET` environment variable+ ## [1.10.0] - 2026-01-06  ### Added
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.10.0+version:            1.11.0 synopsis:   Command-line client for Nix binary cache hosting https://cachix.org
src/Cachix/Client/CNix.hs view
@@ -1,41 +1,131 @@-module Cachix.Client.CNix where+module Cachix.Client.CNix+  ( -- * Store path validation+    StorePathError (..),+    resolveStorePath,+    resolveStorePaths,+    validateStorePath,+    formatStorePathError,+    formatStorePathWarning,+    logStorePathWarning,+    logStorePathWarning', +    -- * Legacy API (deprecated)+    filterInvalidStorePath,+    filterInvalidStorePaths,+    followLinksToStorePath,++    -- * Error handling+    NixError (..),+    catchNixError,+    handleCppExceptions,+  )+where+ import Hercules.CNix.Store (Store, StorePath) import Hercules.CNix.Store qualified as Store import Language.C.Inline.Cpp.Exception import Protolude import System.Console.Pretty (Color (..), Style (..), color, style) --- | Checks whether a store path is valid.-validateStorePath :: Store -> StorePath -> IO (Maybe StorePath)-validateStorePath store storePath = do-  isValid <- Store.isValidPath store storePath `catchNixError` const (return False)-  if isValid-    then return (Just storePath)-    else return Nothing+-- | Error when resolving a store path+data StorePathError+  = -- | Path could not be resolved (doesn't exist, bad symlink, etc.)+    StorePathNotFound Text+  | -- | Path exists but is not valid in the Nix store+    StorePathNotValid+  | -- | Error occurred during validation (e.g., permission denied)+    StorePathError Text+  deriving (Show, Eq) --- | Like 'validateStorePath', but logs a warning when the path is invalid.+-- | Resolve a file path to a validated store path. ----- When validation fails due to an error (e.g., permission denied),--- the actual error message is logged instead of a generic "is not valid" message.-filterInvalidStorePath :: Store -> StorePath -> IO (Maybe StorePath)-filterInvalidStorePath store storePath = do+-- Follows symlinks and validates the resulting store path.+resolveStorePath :: Store -> FilePath -> IO (Either StorePathError StorePath)+resolveStorePath store fp = do+  let pathBytes = encodeUtf8 (toS fp :: Text)+  resolveResult <- tryResolve pathBytes+  case resolveResult of+    Left err -> pure $ Left (StorePathNotFound (msg err))+    Right storePath -> validateStorePath store storePath+  where+    tryResolve pathBytes =+      (Right <$> Store.followLinksToStorePath store pathBytes)+        `catchNixError` (pure . Left)++-- | Resolve multiple file paths to validated store paths.+--+-- Returns a pair of (errors, valid paths). Errors are paired with their+-- original file path for error reporting.+resolveStorePaths :: Store -> [FilePath] -> IO ([(FilePath, StorePathError)], [StorePath])+resolveStorePaths store paths = do+  results <- mapM (resolveStorePath store) paths+  pure $ foldr go ([], []) (zip paths results)+  where+    go (path, Left err) (errs, oks) = ((path, err) : errs, oks)+    go (_, Right sp) (errs, oks) = (errs, sp : oks)++-- | Format a store path error as a reason string (without the path).+formatStorePathError :: StorePathError -> Text+formatStorePathError = \case+  StorePathNotFound reason -> reason+  StorePathNotValid -> "not valid"+  StorePathError reason -> reason++-- | Format a store path error as a user-friendly warning message.+formatStorePathWarning :: FilePath -> StorePathError -> Text+formatStorePathWarning path = \case+  StorePathNotFound reason -> toS path <> ": " <> reason+  StorePathNotValid -> toS path <> " is not valid"+  StorePathError reason -> toS path <> ": " <> reason++-- | Log a store path error as a warning (yellow, to stderr).+logStorePathWarning :: FilePath -> StorePathError -> IO ()+logStorePathWarning path err =+  putErrText $ color Yellow $ "Warning: " <> formatStorePathWarning path err <> ", skipping"++-- | Log a store path validation error as a warning.+--+-- Like 'logStorePathWarning' but for when you have a 'StorePath' instead of a 'FilePath'.+logStorePathWarning' :: Store -> StorePath -> StorePathError -> IO ()+logStorePathWarning' store storePath err = do+  pathBytes <- Store.storePathToPath store storePath+  let path = toS (decodeUtf8With lenientDecode pathBytes :: Text)+  logStorePathWarning path err++-- | Validate that an existing store path is still valid in the Nix store.+--+-- Use this to re-check a previously resolved path, e.g., before pushing+-- a path that may have been garbage collected since it was queued.+validateStorePath :: Store -> StorePath -> IO (Either StorePathError StorePath)+validateStorePath store storePath = do   result <- tryValidate   case result of-    Right True -> return (Just storePath)-    Right False -> do-      logBadStorePath store storePath-      return Nothing+    Left err -> pure $ Left (StorePathError (msg err))+    Right True -> pure $ Right storePath+    Right False -> pure $ Left StorePathNotValid+  where+    tryValidate =+      (Right <$> Store.isValidPath store storePath)+        `catchNixError` (pure . Left)++{-# DEPRECATED filterInvalidStorePath "Use validateStorePath + logStorePathWarning' instead" #-}++-- | Like 'validateStorePath', but logs a warning when the path is invalid.+filterInvalidStorePath :: Store -> StorePath -> IO (Maybe StorePath)+filterInvalidStorePath store storePath =+  validateStorePath store storePath >>= \case+    Right sp -> return (Just sp)     Left err -> do-      logStorePathError store storePath err+      logStorePathWarning' store storePath err       return Nothing-  where-    tryValidate = (Right <$> Store.isValidPath store storePath) `catchNixError` (return . Left) +{-# DEPRECATED filterInvalidStorePaths "Use resolveStorePath + logStorePathWarning instead" #-} filterInvalidStorePaths :: Store -> [StorePath] -> IO [Maybe StorePath] filterInvalidStorePaths store =   traverse (filterInvalidStorePath store) +{-# DEPRECATED followLinksToStorePath "Use resolveStorePath instead" #-}+ -- | Follows all symlinks to a store path, returning the final store path. -- -- Returns Nothing if the path is invalid.@@ -73,19 +163,6 @@   case typ of     Just "nix::BadStorePath" -> logBadPath storePath     _ -> putErrText $ color Red $ style Bold $ "Nix " <> msg---- | Print a warning when a store path is invalid.-logBadStorePath :: Store -> StorePath -> IO ()-logBadStorePath store storePath = do-  path <- Store.storePathToPath store storePath-  logBadPath path---- | Print a warning with the actual error when accessing a store path fails.-logStorePathError :: Store -> StorePath -> NixError -> IO ()-logStorePathError store storePath (NixError {..}) = do-  path <- Store.storePathToPath store storePath-  let pathText = decodeUtf8With lenientDecode path-  putErrText $ color Yellow $ "Warning: " <> pathText <> " - " <> msg <> ", skipping"  -- | Print a warning when the path is invalid. --
src/Cachix/Client/Command/Import.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE OverloadedLabels #-} @@ -25,7 +26,11 @@ import Control.Retry (defaultRetryStatus) import Data.Attoparsec.Text qualified import Data.Conduit.Combinators qualified as C+#if MIN_VERSION_conduit_concurrent_map(0,1,4)+import Data.Conduit.ConcurrentMap (concurrentMapM)+#else import Data.Conduit.ConcurrentMap (concurrentMapM_)+#endif import Data.Conduit.List qualified as CL import Data.Generics.Labels () import Data.Text qualified as T@@ -62,6 +67,8 @@     <&> Amazonka.configureService s3Endpoint       . maybe identity (#region .~) region +-- Disabled due to CPP+{- ORMOLU_DISABLE -} import' :: Env -> PushOptions -> Text -> URI -> IO () import' env pushOptions name s3uri = do   awsEnv <- discoverAwsEnv (URI.getQueryParam s3uri "endpoint") (URI.getQueryParam s3uri "region")@@ -74,7 +81,11 @@         .| CL.concat         .| CL.map (^. object_key)         .| CL.filter (T.isSuffixOf ".narinfo" . Amazonka.Data.Text.toText)+#if MIN_VERSION_conduit_concurrent_map(0,1,4)+        .| concurrentMapM (numJobs pushOptions) (numJobs pushOptions * 2) (uploadNarinfo awsEnv)+#else         .| concurrentMapM_ (numJobs pushOptions) (numJobs pushOptions * 2) (uploadNarinfo awsEnv)+#endif         .| CL.sinkNull   putErrText "All done."   where@@ -154,3 +165,5 @@                              liftIO $ onDone strategy                 | otherwise -> putErrText $ show err++{- ORMOLU_ENABLE -}
src/Cachix/Client/Command/Pin.hs view
@@ -2,7 +2,7 @@  import Cachix.API qualified as API import Cachix.API.Error-import Cachix.Client.CNix (followLinksToStorePath)+import Cachix.Client.CNix (formatStorePathWarning, resolveStorePath) import Cachix.Client.Config qualified as Config import Cachix.Client.Env (Env (..)) import Cachix.Client.Exception (CachixException (..))@@ -21,8 +21,12 @@ pin env pinOpts = do   authToken <- Config.getAuthTokenRequired (config env)   storePath <- withStore $ \store -> do-    mpath <- followLinksToStorePath store (encodeUtf8 $ pinStorePath pinOpts)-    maybe exitFailure (storePathToPath store) mpath+    let filePath = toS (pinStorePath pinOpts)+    resolveStorePath store filePath >>= \case+      Left err -> do+        putErrText $ formatStorePathWarning filePath err+        exitFailure+      Right sp -> storePathToPath store sp   traverse_ (validateArtifact (toS storePath)) (pinArtifacts pinOpts)   let pinCreate =         PinCreate.PinCreate
src/Cachix/Client/Command/Push.hs view
@@ -12,7 +12,7 @@ where  import Cachix.API qualified as API-import Cachix.Client.CNix (filterInvalidStorePath, followLinksToStorePath)+import Cachix.Client.CNix (logStorePathWarning, resolveStorePaths) import Cachix.Client.Config qualified as Config import Cachix.Client.Env (Env (..)) import Cachix.Client.Exception (CachixException (..))@@ -25,7 +25,6 @@ import Cachix.Types.BinaryCache (BinaryCacheName) import Cachix.Types.BinaryCache qualified as BinaryCache import Control.Exception.Safe (throwM)-import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Control.Retry (RetryStatus (rsIterNumber)) import Data.ByteString qualified as BS import Data.Conduit qualified as Conduit@@ -59,17 +58,14 @@       -- This is somewhat like the behavior of `cat` for example.       (_, paths) -> return paths   withPushParams env opts name $ \pushParams -> do-    normalized <- liftIO $-      for inputStorePaths $ \path ->-        runMaybeT $ do-          storePath <- MaybeT $ followLinksToStorePath (pushParamsStore pushParams) (encodeUtf8 path)-          MaybeT $ filterInvalidStorePath (pushParamsStore pushParams) storePath+    (errors, validPaths) <- liftIO $ resolveStorePaths (pushParamsStore pushParams) (map toS inputStorePaths)+    liftIO $ forM_ errors $ uncurry logStorePathWarning     pushedPaths <-       pushClosure         (mapConcurrentlyBounded (numJobs opts))         pushParams-        (catMaybes normalized)-    case (length normalized, length pushedPaths) of+        validPaths+    case (length inputStorePaths, length pushedPaths) of       (0, _) -> putErrText "Nothing to push."       (_, 0) -> putErrText "Nothing to push - all store paths are already on Cachix."       _ -> putErrText "\nAll done."
src/Cachix/Client/Command/Watch.hs view
@@ -13,6 +13,8 @@   ( DaemonOptions (..),     PushOptions (..),     WatchExecMode (..),+    defaultKeepAliveInterval,+    defaultKeepAliveTimeout,   ) import Cachix.Client.Push import Cachix.Client.WatchStore qualified as WatchStore@@ -34,7 +36,7 @@ import Protolude.Conv import Servant.Conduit () import System.Console.Pretty-import System.Environment (getEnvironment)+import System.Environment (getEnvironment, lookupEnv) import System.IO.Error (isEOFError) import System.IO.Temp (withTempFile) import System.Posix.Signals qualified as Signals@@ -72,15 +74,18 @@ -- -- Requires the user to be a trusted user in a multi-user installation. watchExecDaemon :: Env -> PushOptions -> NarinfoQueryOptions -> BinaryCacheName -> Text -> [Text] -> IO ()-watchExecDaemon env pushOpts batchOptions cacheName cmd args =-  Daemon.PostBuildHook.withSetup Nothing $ \hookEnv ->+watchExecDaemon env pushOpts batchOptions cacheName cmd args = do+  envSocketPath <- lookupEnv "CACHIX_DAEMON_SOCKET"+  Daemon.PostBuildHook.withSetup envSocketPath $ \hookEnv ->     withTempFile (Daemon.PostBuildHook.tempDir hookEnv) "daemon-log-capture" $ \_ logHandle ->       withStore $ \store -> do         let daemonOptions =               DaemonOptions                 { daemonAllowRemoteStop = False,                   daemonNarinfoQueryOptions = batchOptions,-                  daemonSocketPath = Just (Daemon.PostBuildHook.daemonSock hookEnv)+                  daemonSocketPath = Just (Daemon.PostBuildHook.daemonSock hookEnv),+                  daemonKeepAliveInterval = defaultKeepAliveInterval,+                  daemonKeepAliveTimeout = defaultKeepAliveTimeout                 }         daemon <- Daemon.new env store daemonOptions (Just logHandle) pushOpts cacheName
src/Cachix/Client/Exception.hs view
@@ -22,6 +22,7 @@   | BinaryCacheNotFound Text   | ImportUnsupportedHash Text   | RemoveCacheUnsupported Text+  | InvalidStorePath Text   deriving (Show, Typeable)  instance Exception CachixException where@@ -44,3 +45,4 @@   displayException (BinaryCacheNotFound s) = toS s   displayException (ImportUnsupportedHash s) = toS s   displayException (RemoveCacheUnsupported s) = toS s+  displayException (InvalidStorePath s) = toS s
src/Cachix/Client/OptionsParser.hs view
@@ -14,6 +14,8 @@     defaultChunkSize,     defaultNumJobs,     defaultOmitDeriver,+    defaultKeepAliveInterval,+    defaultKeepAliveTimeout,      -- * Watch exec     WatchExecMode (..),@@ -156,6 +158,12 @@ defaultOmitDeriver :: Bool defaultOmitDeriver = False +defaultKeepAliveInterval :: Int+defaultKeepAliveInterval = 30++defaultKeepAliveTimeout :: Int+defaultKeepAliveTimeout = 180+ defaultPushOptions :: PushOptions defaultPushOptions =   PushOptions@@ -178,7 +186,9 @@ data DaemonOptions = DaemonOptions   { daemonAllowRemoteStop :: Bool,     daemonNarinfoQueryOptions :: NarinfoQueryOptions,-    daemonSocketPath :: Maybe FilePath+    daemonSocketPath :: Maybe FilePath,+    daemonKeepAliveInterval :: Int,+    daemonKeepAliveTimeout :: Int   }   deriving (Show) @@ -558,6 +568,8 @@     <$> remoteStopOption     <*> batchConfigParser     <*> socketOption+    <*> keepAliveIntervalOption+    <*> keepAliveTimeoutOption   where     socketOption =       optional . strOption $@@ -568,6 +580,22 @@      remoteStopOption =       enableDisableFlag True "remote-stop" "the remote stop command which allows clients to remotely shut down the daemon. Remote stop should be disabled in environments where the lifecycle of the daemon is handled by a service manager, like systemd."++    keepAliveIntervalOption =+      option auto $+        long "keep-alive-interval"+          <> metavar "SECONDS"+          <> help "Keep-alive ping interval in seconds (0 to disable)"+          <> value defaultKeepAliveInterval+          <> showDefault++    keepAliveTimeoutOption =+      option auto $+        long "keep-alive-timeout"+          <> metavar "SECONDS"+          <> help "Keep-alive timeout in seconds before considering the daemon stalled"+          <> value defaultKeepAliveTimeout+          <> showDefault  batchConfigParser :: Parser NarinfoQueryOptions batchConfigParser =
src/Cachix/Client/Push.hs view
@@ -46,6 +46,7 @@ import Cachix.API qualified as API import Cachix.API.Error import Cachix.API.Signing (fingerprint, passthroughHashSink, passthroughHashSinkB16, passthroughSizeSink)+import Cachix.Client.CNix (formatStorePathWarning, validateStorePath) import Cachix.Client.Exception (CachixException (..)) import Cachix.Client.Push.S3 qualified as Push.S3 import Cachix.Client.Retry (retryAll, retryHttp)@@ -211,12 +212,14 @@   let store = pushParamsStore pushParams       strategy = pushParamsStrategy pushParams storePath -  -- TODO: storePathText is redundant. Use storePath directly.   storePathText <- liftIO $ Store.storePathToPath store storePath+  let path = toS storePathText -  -- This should be a noop because storePathText came from a StorePath-  normalized <- liftIO $ Store.followLinksToStorePath store storePathText-  pathInfo <- newPathInfoFromStorePath store normalized+  -- Validate the path is still valid (may have been GC'd since queued)+  liftIO (validateStorePath store storePath) >>= \case+    Right _ -> pure ()+    Left err -> throwM $ InvalidStorePath $ formatStorePathWarning path err+  pathInfo <- newPathInfoFromStorePath store storePath   let narSize = fromIntegral (piNarSize pathInfo)    onAttempt strategy retrystatus narSize
src/Cachix/Client/Push/S3.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE Rank2Types #-} @@ -23,7 +24,11 @@ import Data.Conduit (ConduitT, handleC, (.|)) import Data.Conduit.ByteString (ChunkSize, chunkStream) import Data.Conduit.Combinators qualified as CC+#if MIN_VERSION_conduit_concurrent_map(0,1,4)+import Data.Conduit.ConcurrentMap (concurrentMapM)+#else import Data.Conduit.ConcurrentMap (concurrentMapM_)+#endif import Data.List (lookup) import Data.List.NonEmpty qualified as NonEmpty import Data.UUID (UUID)@@ -49,6 +54,8 @@   }   deriving stock (Eq, Show) +-- Disabled due to CPP+{- ORMOLU_DISABLE -} uploadMultipart ::   forall m.   (MonadUnliftIO m, MonadResource m) =>@@ -67,7 +74,11 @@     Right (Multipart.CreateMultipartUploadResponse {narId, uploadId}) -> do       handleC (abortMultipartUpload narId uploadId) $         chunkStream (Just (chunkSize options))+#if MIN_VERSION_conduit_concurrent_map(0,1,4)+          .| concurrentMapM (numConcurrentChunks options) outputBufferSize (uploadPart narId uploadId)+#else           .| concurrentMapM_ (numConcurrentChunks options) outputBufferSize (uploadPart narId uploadId)+#endif           .| completeMultipartUpload narId uploadId   where     -- The size of the temporary output buffer.@@ -124,3 +135,5 @@       let abortMultipartUploadRequest = API.abortMultipartUpload cachixClient authToken cacheName narId uploadId       _ <- liftIO $ retryHttp $ withClientM abortMultipartUploadRequest env escalate       return $ Left err++{- ORMOLU_ENABLE -}
src/Cachix/Client/PushQueue.hs view
@@ -12,7 +12,7 @@   ) where -import Cachix.Client.CNix (filterInvalidStorePath)+import Cachix.Client.CNix (logStorePathWarning', validateStorePath) import Cachix.Client.Push (getMissingPathsForClosure) import Cachix.Client.Push qualified as Push import Cachix.Client.Retry (retryAll)@@ -44,11 +44,13 @@ worker pushParams workerState = forever $ do   storePath <- atomically $ TBQueue.readTBQueue $ pushQueue workerState   bracket_ (inProgressModify (+ 1)) (inProgressModify (\x -> x - 1)) $ do-    maybeStorePath <- filterInvalidStorePath (Push.pushParamsStore pushParams) storePath-    for_ maybeStorePath $ \validStorePath -> do-      (_, missingPaths) <- getMissingPathsForClosure pushParams [validStorePath]-      for_ missingPaths $ \missingPath ->-        retryAll $ Push.uploadStorePath pushParams missingPath+    let store = Push.pushParamsStore pushParams+    validateStorePath store storePath >>= \case+      Left err -> logStorePathWarning' store storePath err+      Right validStorePath -> do+        (_, missingPaths) <- getMissingPathsForClosure pushParams [validStorePath]+        for_ missingPaths $ \missingPath ->+          retryAll $ Push.uploadStorePath pushParams missingPath   where     inProgressModify f =       atomically $ modifyTVar' (inProgress workerState) f
src/Cachix/Client/WatchStore.hs view
@@ -3,12 +3,11 @@   ) where -import Cachix.Client.CNix (filterInvalidStorePath)+import Cachix.Client.CNix (logStorePathWarning, resolveStorePath) import Cachix.Client.Push import Cachix.Client.PushQueue qualified as PushQueue import Control.Concurrent.STM.TBQueue qualified as TBQueue import Hercules.CNix.Store (Store)-import Hercules.CNix.Store qualified as Store import Protolude import System.FSNotify import System.Systemd.Daemon qualified as Systemd@@ -25,10 +24,10 @@  queueStorePathAction :: Store -> PushQueue.Queue -> Event -> IO () queueStorePathAction store queue (Removed lockFile _ _) = do-  sp <- Store.parseStorePath store (encodeUtf8 $ toS $ dropLast 5 lockFile)-  filterInvalidStorePath store sp >>= \case-    Nothing -> return ()-    Just p -> atomically $ TBQueue.writeTBQueue queue p+  let filePath = dropLast 5 lockFile+  resolveStorePath store filePath >>= \case+    Left err -> logStorePathWarning filePath err+    Right p -> atomically $ TBQueue.writeTBQueue queue p queueStorePathAction _ _ _ = return ()  dropLast :: Int -> [a] -> [a]
src/Cachix/Daemon.hs view
@@ -143,7 +143,7 @@    eventLoopRes <- EventLoop.run daemonEventLoop $ \case     AddSocketClient conn ->-      SocketStore.addSocket conn (Listen.handleClient daemonEventLoop) daemonClients+      SocketStore.addSocket conn (Listen.handleClient daemonEventLoop daemonClients) daemonClients     RemoveSocketClient socketId ->       SocketStore.removeSocket socketId daemonClients     ReconnectSocket ->@@ -245,7 +245,11 @@      -- SIGINT: First try to shutdown gracefully, on second press force exit     intHandler :: ThreadId -> IORef Bool -> Daemon ()-    intHandler mainThreadId interruptRef = do+    intHandler mainThreadId interruptRef =+      handleInterrupt mainThreadId interruptRef++    handleInterrupt :: ThreadId -> IORef Bool -> Daemon ()+    handleInterrupt mainThreadId interruptRef = do       liftIO CNix.Util.triggerInterrupt       isSecondInterrupt <- liftIO $ atomicModifyIORef' interruptRef (True,)       eventLoop <- asks daemonEventLoop@@ -256,6 +260,8 @@           startExitTimer mainThreadId           -- Force shutdown at the event loop level to ensure exit even if queue is full           EventLoop.exitLoopWithFailure EventLoopClosed eventLoop+          -- Interrupt the main thread right away so we don't wait for graceful shutdown.+          liftIO $ throwTo mainThreadId ExitSuccess         else do           Katip.logFM Katip.InfoS "Shutting down gracefully (Ctrl+C again to force exit)..."           EventLoop.send eventLoop ShutdownGracefully@@ -323,16 +329,19 @@ shutdownGracefully = do   DaemonEnv {..} <- ask -  -- Stop the narinfo batch processor first (before closing the queue)+  -- 1. Drain: set latch (reject new jobs), wait for in-flight jobs to complete+  drainWithLogging daemonPushManager++  -- 2. Stop the batch processor (safe: all jobs completed the full pipeline)   liftIO $ PushManager.stopBatchProcessor daemonPushManager -  -- Stop the push manager and wait for any remaining paths to be uploaded-  shutdownPushManager daemonPushManager+  -- 3. Close the task queue+  liftIO $ PushManager.closePushManager daemonPushManager -  -- Stop worker threads+  -- 4. Stop worker threads (workers see closed queue and exit)   withTakeMVar daemonWorkerThreads Worker.stopWorkers -  -- Close all event subscriptions+  -- 5. Close all event subscriptions   withTakeMVar daemonSubscriptionManagerThread (shutdownSubscriptions daemonSubscriptionManager)    failedJobs <-@@ -342,27 +351,28 @@           then Right ()           else Left DaemonPushFailure -  -- Gracefully close open connections to clients-  Async.mapConcurrently_ (sayGoodbye pushResult) =<< SocketStore.toList daemonClients+  -- 6. Gracefully close open connections to clients+  Async.mapConcurrently_ (sayGoodbye daemonClients pushResult) =<< SocketStore.toList daemonClients    return pushResult   where-    shutdownPushManager daemonPushManager = do+    drainWithLogging daemonPushManager = do       queuedStorePathCount <- PushManager.runPushManager daemonPushManager 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+      Katip.logFM Katip.DebugS "Waiting for push manager to drain remaining 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."+      drained <- liftIO $ PushManager.drainPushManager timeoutOptions daemonPushManager+      if drained+        then Katip.logFM Katip.DebugS "Push manager drained."+        else Katip.logFM Katip.WarningS "Push manager drain timed out. Some jobs may not have completed."      shutdownSubscriptions daemonSubscriptionManager subscriptionManagerThread = do       Katip.logFM Katip.DebugS "Shutting down event manager..."@@ -370,13 +380,14 @@       _ <- Async.wait subscriptionManagerThread       Katip.logFM Katip.DebugS "Event manager shut down." -    sayGoodbye exitResult socket = do+    sayGoodbye socketStore exitResult socket = do       let clientSock = SocketStore.socket socket       let clientThread = SocketStore.handlerThread socket+      let clientSocketId = SocketStore.socketId socket       Async.cancel clientThread        -- Wave goodbye to the client that requested the shutdown-      liftIO $ Listen.serverBye clientSock exitResult+      liftIO $ Listen.serverBye clientSocketId socketStore exitResult       liftIO $ Socket.shutdown clientSock Socket.ShutdownBoth `catchAny` (\_ -> return ())       -- Wait for the other end to disconnect       ebs <- liftIO $ try $ Socket.BS.recv clientSock 4096
src/Cachix/Daemon/Client.hs view
@@ -1,5 +1,6 @@ module Cachix.Daemon.Client (push, stop) where +import Cachix.Client.CNix (logStorePathWarning, resolveStorePaths) import Cachix.Client.Env as Env import Cachix.Client.Exception (CachixException (..)) import Cachix.Client.OptionsParser (DaemonOptions (..), DaemonPushOptions (..))@@ -43,32 +44,37 @@     SocketDecodingError err -> "Failed to decode the message from socket: " <> toS err  -- | Run socket communication with ping/pong handling-withSocketComm :: Socket.Socket -> (STM (Maybe (Either SocketError Protocol.DaemonMessage)) -> (Protocol.ClientMessage -> STM ()) -> IO a) -> IO a-withSocketComm sock action = do-  let size = 1000-  (rx, tx) <- atomically $ (,) <$> newTBMQueue size <*> newTBMQueue size+withSocketComm :: DaemonOptions -> Socket.Socket -> (STM (Maybe (Either SocketError Protocol.DaemonMessage)) -> (Protocol.ClientMessage -> STM ()) -> IO a) -> IO a+withSocketComm daemonOptions sock action = do+  let size = 5000+  let prioritySize = 200+  (rxPriority, rx, tx) <- atomically $ (,,) <$> newTBMQueue prioritySize <*> newTBMQueue size <*> newTBMQueue size    lastPongRef <- newIORef =<< getCurrentTime-  rxThread <- Async.async (handleIncoming lastPongRef rx sock)+  rxThread <- Async.async (handleIncoming lastPongRef rxPriority rx sock)   txThread <- Async.async (handleOutgoing tx sock)-  pingThread <- Async.async (runPingThread lastPongRef rx tx)+  pingThread <-+    if daemonKeepAliveInterval daemonOptions > 0+      then Just <$> Async.async (runPingThread lastPongRef rxPriority tx)+      else pure Nothing -  let threads = [rxThread, txThread, pingThread]+  let threads = rxThread : txThread : maybeToList pingThread   mapM_ Async.link threads -  finally (action (readTBMQueue rx) (writeTBMQueue tx)) (mapM_ Async.cancel threads)+  let receive = readTBMQueue rxPriority `orElse` readTBMQueue rx+  finally (action receive (writeTBMQueue tx)) (mapM_ Async.cancel threads)   where-    runPingThread lastPongRef rx tx = go+    runPingThread lastPongRef rxPriority tx = go       where         go = do           timestamp <- getCurrentTime           lastPong <- readIORef lastPongRef -          if timestamp >= addUTCTime 20 lastPong-            then atomically $ writeTBMQueue rx (Left SocketStalled)+          if timestamp >= addUTCTime (fromIntegral (daemonKeepAliveTimeout daemonOptions)) lastPong+            then atomically $ writeTBMQueue rxPriority (Left SocketStalled)             else do               atomically $ writeTBMQueue tx Protocol.ClientPing-              threadDelay (2 * 1000 * 1000)+              threadDelay (daemonKeepAliveInterval daemonOptions * 1000 * 1000)               go      handleOutgoing tx sock' = go@@ -81,9 +87,9 @@               Retry.retryAll $ const $ Socket.LBS.sendAll sock' $ Protocol.newMessage msg               go -    handleIncoming lastPongRef rx sock' = go BS.empty+    handleIncoming lastPongRef rxPriority rx sock' = go BS.empty       where-        socketClosed = atomically $ writeTBMQueue rx (Left SocketClosed)+        socketClosed = atomically $ writeTBMQueue rxPriority (Left SocketClosed)          go leftovers = do           ebs <- liftIO $ try $ Socket.BS.recv sock' 4096@@ -100,12 +106,16 @@                   Left err -> do                     let terr = toS err                     putErrText terr-                    atomically $ writeTBMQueue rx (Left (SocketDecodingError terr))+                    atomically $ writeTBMQueue rxPriority (Left (SocketDecodingError terr))                   Right msg -> do                     case msg of                       Protocol.DaemonPong -> do                         writeIORef lastPongRef =<< getCurrentTime-                        atomically $ writeTBMQueue rx (Right msg)+                        atomically $ writeTBMQueue rxPriority (Right msg)+                      Protocol.DaemonExit {} ->+                        atomically $ writeTBMQueue rxPriority (Right msg)+                      Protocol.DaemonPushEvent PushEvent {eventMessage = PushStorePathProgress {}} ->+                        void $ atomically $ tryWriteTBMQueue rx (Right msg)                       _ -> atomically $ writeTBMQueue rx (Right msg)                go newLeftovers@@ -123,8 +133,9 @@       (_, paths) -> return paths    storePaths <- Store.withStore $ \store -> do-    inputStorePaths' <- mapM (Store.followLinksToStorePath store . BS8.pack) inputStorePaths-    mapM (fmap (toS . BS8.unpack) . Store.storePathToPath store) inputStorePaths'+    (errors, validPaths) <- resolveStorePaths store inputStorePaths+    for_ errors $ uncurry logStorePathWarning+    mapM (fmap (toS . BS8.unpack) . Store.storePathToPath store) validPaths    withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do     let shouldWait = Options.shouldWait daemonPushOptions@@ -133,7 +144,7 @@     Socket.LBS.sendAll sock $ Protocol.newMessage pushRequest     unless shouldWait exitSuccess -    withSocketComm sock $ \receive _send -> do+    withSocketComm daemonOptions sock $ \receive _send -> do       progressState <- Daemon.Progress.newProgressState       failureRef <- newIORef False @@ -167,7 +178,7 @@ stop :: Env -> DaemonOptions -> IO () stop _env daemonOptions =   withDaemonConn (daemonSocketPath daemonOptions) $ \sock -> do-    withSocketComm sock $ \receive send -> do+    withSocketComm daemonOptions sock $ \receive send -> do       atomically $ send Protocol.ClientStop        fix $ \loop -> do
src/Cachix/Daemon/Listen.hs view
@@ -11,6 +11,7 @@ import Cachix.Client.Config.Orphans () import Cachix.Daemon.EventLoop qualified as EventLoop import Cachix.Daemon.Protocol as Protocol+import Cachix.Daemon.SocketStore qualified as SocketStore import Cachix.Daemon.Types   ( DaemonError,     DaemonEvent@@ -21,7 +22,7 @@     toExitCodeInt,   ) import Cachix.Daemon.Types.EventLoop (EventLoop)-import Cachix.Daemon.Types.SocketStore (SocketId)+import Cachix.Daemon.Types.SocketStore (SocketId, SocketStore) import Control.Exception.Safe (catchAny) import Control.Monad.Catch qualified as E import Data.Aeson qualified as Aeson@@ -30,7 +31,6 @@ import Network.Socket (Socket) import Network.Socket qualified as Socket import Network.Socket.ByteString qualified as Socket.BS-import Network.Socket.ByteString.Lazy qualified as Socket.LBS import Protolude import System.Directory   ( XdgDirectory (..),@@ -76,10 +76,11 @@   forall m a.   (E.MonadMask m, Katip.KatipContext m) =>   EventLoop DaemonEvent a ->+  SocketStore ->   SocketId ->   Socket ->   m ()-handleClient eventloop socketId conn = do+handleClient eventloop socketStore socketId conn = do   go BS.empty `E.finally` removeClient   where     go leftovers = do@@ -98,7 +99,7 @@             EventLoop.send eventloop (ReceivedMessage socketId msg)             case msg of               Protocol.ClientPing ->-                liftIO $ Socket.LBS.sendAll conn $ Protocol.newMessage DaemonPong+                liftIO $ SocketStore.sendAll socketId (Protocol.newMessage DaemonPong) socketStore               _ -> return ()            go newLeftovers@@ -114,9 +115,9 @@       return Nothing     Right msg -> return (Just msg) -serverBye :: Socket.Socket -> Either DaemonError () -> IO ()-serverBye sock exitResult =-  Socket.LBS.sendAll sock (Protocol.newMessage (DaemonExit exitStatus)) `catchAny` (\_ -> return ())+serverBye :: SocketId -> SocketStore -> Either DaemonError () -> IO ()+serverBye socketId socketStore exitResult =+  SocketStore.sendAll socketId (Protocol.newMessage (DaemonExit exitStatus)) socketStore `catchAny` (\_ -> return ())   where     exitStatus = DaemonExitStatus {exitCode, exitMessage}     exitCode = toExitCodeInt exitResult
src/Cachix/Daemon/NarinfoQuery.hs view
@@ -253,6 +253,10 @@   let delayMicros = ceiling (delay * 1000000) -- Convert to microseconds   registerDelay delayMicros +-- | Check if there are pending requests waiting to be processed+hasPendingRequests :: QueryState requestId -> Bool+hasPendingRequests = not . Seq.null . qsPendingRequests+ -- | Check if the current query has timed out isQueryTimedOut :: QueryState requestId -> STM Bool isQueryTimedOut queryState =@@ -276,46 +280,42 @@ waitForBatchOrShutdown config stateVar = do   queryState <- readTVar stateVar -  -- Check for shutdown first+  let extractBatch = do+        let requests = qsPendingRequests queryState+            allPaths = Set.toList (qsAccumulatedPaths queryState)+            startTime = qsQueryStartTime queryState+        writeTVar stateVar $+          queryState+            { qsPendingRequests = Seq.empty,+              qsAccumulatedPaths = Set.empty,+              qsQueryStartTime = Nothing,+              qsTimeoutVar = Nothing+            }+        return $+          Just+            ReadyBatch+              { rbRequests = requests,+                rbAllPaths = allPaths,+                rbBatchStartTime = startTime+              }+   if not (qsRunning queryState)-    then return Nothing+    then+      -- Shutting down: flush remaining requests as a final batch, or signal done+      if hasPendingRequests queryState then extractBatch else return Nothing     else do-      -- Check if we have any pending requests-      if Seq.null (qsPendingRequests queryState)+      if not (hasPendingRequests queryState)         then retry -- No work, wait for requests         else do           -- We have requests, check if batch is ready           let pathCount = Set.size (qsAccumulatedPaths queryState)               sizeReady = pathCount >= nqoMaxBatchSize config-              -- If timeout is 0, process immediately               immediateMode = nqoMaxWaitTime config <= 0 -          -- Check timeout condition           timeoutReady <- isQueryTimedOut queryState            if sizeReady || timeoutReady || immediateMode-            then do-              -- Batch is ready, extract data and clear state-              let requests = qsPendingRequests queryState-                  allPaths = Set.toList (qsAccumulatedPaths queryState)-                  startTime = qsQueryStartTime queryState--              -- Clear the batch state-              writeTVar stateVar $-                queryState-                  { qsPendingRequests = Seq.empty,-                    qsAccumulatedPaths = Set.empty,-                    qsQueryStartTime = Nothing,-                    qsTimeoutVar = Nothing-                  }--              return $-                Just-                  ReadyBatch-                    { rbRequests = requests,-                      rbAllPaths = allPaths,-                      rbBatchStartTime = startTime-                    }+            then extractBatch             else retry -- Not ready yet, wait for timeout or more requests  -- | Time refresh thread that updates cached time every few seconds
src/Cachix/Daemon/PostBuildHook.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE TypeApplications #-} @@ -13,10 +14,13 @@     -- * Internal     buildNixConfEnv,     buildNixUserConfFilesEnv,+    withRunnerFriendlyTempDirectory,+    unixSocketMaxPath,   ) where  import Control.Monad.Catch (MonadMask)+import Control.Monad.Catch qualified as E import Data.Containers.ListUtils (nubOrd) import Data.String (String) import Data.String.Here@@ -26,11 +30,13 @@     XdgDirectoryList (XdgConfigDirs),     getXdgDirectory,     getXdgDirectoryList,+    removeDirectoryRecursive,   ) import System.Environment (getExecutablePath, lookupEnv) import System.FilePath ((</>))-import System.IO.Temp (getCanonicalTemporaryDirectory, withTempDirectory)+import System.IO.Temp (getCanonicalTemporaryDirectory) import System.Posix.Files+import System.Posix.Temp (mkdtemp)  type EnvVar = (String, String) @@ -141,14 +147,37 @@   $OUT_PATHS   |] +-- | Maximum length for Unix socket paths (sizeof(sockaddr_un.sun_path)).+-- Linux: 108, macOS/BSD: 104.+unixSocketMaxPath :: Int+#if defined(linux_HOST_OS)+unixSocketMaxPath = 108+#else+unixSocketMaxPath = 104+#endif+ -- | 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.+-- Uses RUNNER_TEMP if set and the resulting socket path fits within Unix limits.+-- Falls back to system temp directory if the path would be too long. -- The directory is deleted after use. withRunnerFriendlyTempDirectory :: (MonadIO m, MonadMask m) => String -> (FilePath -> m a) -> m a-withRunnerFriendlyTempDirectory name action = do+withRunnerFriendlyTempDirectory prefix action = do   runnerTempDir <- liftIO $ lookupEnv "RUNNER_TEMP"   systemTempDir <- liftIO getCanonicalTemporaryDirectory-  let tempDir = maybe systemTempDir toS runnerTempDir-  withTempDirectory tempDir name action++  -- Check if RUNNER_TEMP would result in a path that's too long for Unix sockets.+  -- mkdtemp appends 6 random chars to the prefix.+  let wouldFit dir = length (dir </> prefix <> "XXXXXX" </> "daemon.sock") < unixSocketMaxPath+      baseDir = case runnerTempDir of+        Just rt | wouldFit rt -> rt+        _ -> systemTempDir++  let dirTemplate = baseDir </> prefix+  E.bracket+    (liftIO $ mkdtemp dirTemplate)+    (liftIO . ignoringIOErrors . removeDirectoryRecursive)+    action+  where+    ignoringIOErrors :: IO () -> IO ()+    ignoringIOErrors io = io `E.catch` (\(_ :: IOException) -> return ())
src/Cachix/Daemon/Progress.hs view
@@ -31,11 +31,13 @@   = ProgressBar       { path :: String,         size :: Int64,-        progressBar :: Ascii.ProgressBar+        progressBar :: Ascii.ProgressBar,+        lastBytesRef :: IORef Int64       }   | FallbackText       { path :: String,-        size :: Int64+        size :: Int64,+        lastBytesRef :: IORef Int64       }  -- | State for tracking multiple progress bars@@ -47,6 +49,7 @@  new :: Handle -> String -> Int64 -> PushRetryStatus -> IO UploadProgress new hdl path size retryStatus = do+  lastBytesRef <- newIORef 0   isCI <- liftIO $ (== Just "true") <$> lookupEnv "CI"   isTerminal <- liftIO $ hIsTerminalDevice hdl   if isTerminal && not isCI@@ -65,6 +68,13 @@ tick ProgressBar {..} deltaBytes = Ascii.tickN progressBar (fromIntegral deltaBytes) tick _ _ = pure () +tickAbsolute :: UploadProgress -> Int64 -> IO ()+tickAbsolute progress currentBytes = do+  lastBytes <- readIORef (lastBytesRef progress)+  let delta = max 0 (currentBytes - lastBytes)+  when (delta > 0) $ tick progress delta+  writeIORef (lastBytesRef progress) currentBytes+ fail :: UploadProgress -> IO () fail ProgressBar {..} =   clearAsciiWith progressBar (uploadFailed path)@@ -195,10 +205,10 @@         Nothing -> do           progress <- new stderr (toS path) pathSize retryStatus           writeIORef statsRef $ HashMap.insert path (progress, retryStatus) stats-    PushStorePathProgress path _ newBytes -> do+    PushStorePathProgress path currentBytes _deltaBytes -> do       case HashMap.lookup path stats of         Nothing -> return ()-        Just (progress, _) -> tick progress newBytes+        Just (progress, _) -> tickAbsolute progress currentBytes     PushStorePathDone path -> do       mapM_ (complete . fst) (HashMap.lookup path stats)     PushStorePathFailed path _ -> do
src/Cachix/Daemon/PushManager.hs view
@@ -1,7 +1,8 @@ module Cachix.Daemon.PushManager   ( newPushManagerEnv,     runPushManager,-    stopPushManager,+    drainPushManager,+    closePushManager,      -- * Push strategy     newPushStrategy,@@ -45,7 +46,7 @@   ) where -import Cachix.Client.CNix (filterInvalidStorePath, followLinksToStorePath)+import Cachix.Client.CNix (formatStorePathError, logStorePathWarning, resolveStorePaths) import Cachix.Client.Command.Push hiding (pushStrategy) import Cachix.Client.OptionsParser as Client.OptionsParser   ( PushOptions (..),@@ -55,6 +56,7 @@ import Cachix.Daemon.NarinfoQuery qualified as NarinfoQuery import Cachix.Daemon.Protocol qualified as Protocol import Cachix.Daemon.PushManager.PushJob qualified as PushJob+import Cachix.Daemon.ShutdownLatch qualified as ShutdownLatch import Cachix.Daemon.TaskQueue import Cachix.Daemon.Types.Log (Logger) import Cachix.Daemon.Types.PushEvent (PushEvent (..), PushEventMessage (..), newPushRetryStatus)@@ -74,6 +76,7 @@ import Data.Set qualified as Set import Data.Text qualified as T import Data.Time (UTCTime, diffUTCTime, getCurrentTime, secondsToNominalDiffTime)+import GHC.Clock (getMonotonicTimeNSec) import Hercules.CNix (StorePath) import Hercules.CNix.Store (Store, parseStorePath, storePathToPath) import Katip qualified@@ -92,23 +95,33 @@   pmTaskQueue <- atomically newTaskQueue   pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)   pmLastEventTimestamp <- newTVarIO =<< getCurrentTime+  let pmProgressEmitIntervalNs = 200 * 1000 * 1000   let pmOnPushEvent id pushEvent = updateTimestampTVar pmLastEventTimestamp >> onPushEvent id pushEvent    -- Create query manager with callback that queues ProcessQueryResponse tasks   let batchCallback requestId response = do         atomically $ writeTask pmTaskQueue $ HandleMissingPathsResponse requestId response   pmNarinfoQueryManager <- NarinfoQuery.new batchOptions batchCallback+  pmShutdownLatch <- ShutdownLatch.newShutdownLatch    return $ PushManagerEnv {..}  runPushManager :: (MonadIO m) => PushManagerEnv -> PushManager a -> m a runPushManager env f = liftIO $ unPushManager f `runReaderT` env -stopPushManager :: TimeoutOptions -> PushManagerEnv -> IO ()-stopPushManager timeoutOptions PushManagerEnv {..} = do+-- | Set the shutdown latch (rejecting new jobs), then wait for all+-- in-flight jobs to complete with a timeout.+-- Returns True if all jobs completed, False if timed out.+drainPushManager :: TimeoutOptions -> PushManagerEnv -> IO Bool+drainPushManager timeoutOptions PushManagerEnv {..} = do+  ShutdownLatch.initiateShutdown () pmShutdownLatch   atomicallyWithTimeout timeoutOptions pmLastEventTimestamp $ do     pendingJobs <- readTVar pmPendingJobCount     check (pendingJobs <= 0)++-- | Close the task queue so workers see it as closed and exit.+closePushManager :: PushManagerEnv -> IO ()+closePushManager PushManagerEnv {..} =   atomically $ closeTaskQueue pmTaskQueue  -- | Start the batch processor for narinfo queries@@ -141,12 +154,16 @@   Katip.logLocM Katip.DebugS $ Katip.ls $ "Queued push job " <> (show pushId :: Text)    let queueJob = do-        modifyTVar' pmPushJobs $ HashMap.insert pushId pushJob-        incrementTVar pmPendingJobCount-        res <- tryWriteTask pmTaskQueue $ QueryMissingPaths pushId-        case res of-          Just True -> return True-          _ -> return False+        shuttingDown <- ShutdownLatch.isShuttingDownSTM pmShutdownLatch+        if shuttingDown+          then return False+          else do+            modifyTVar' pmPushJobs $ HashMap.insert pushId pushJob+            incrementTVar pmPendingJobCount+            res <- tryWriteTask pmTaskQueue $ QueryMissingPaths pushId+            case res of+              Just True -> return True+              _ -> return False    didQueue <- liftIO $ atomically queueJob @@ -280,6 +297,10 @@    withPushJob pushId $ \pushJob -> do     pushStarted pushJob+    -- Emit PushStorePathSkipped events for paths already present in the cache+    let skippedPaths = Set.difference (PushJob.rcAllPaths closure) (PushJob.rcMissingPaths closure)+    forM_ skippedPaths $ \path ->+      sendStorePathEvent [pushId] (PushStorePathSkipped path)     -- Create STM action for each path and then run everything atomically     queueStorePaths pushId $ Set.toList (PushJob.rcMissingPaths closure)     -- Check if the job is already completed, i.e. all paths have been skipped.@@ -329,8 +350,17 @@       withPushJob pushId $ \pushJob -> do         let sps = Protocol.storePaths (pushRequest pushJob)             store = pushParamsStore pushParams-        validPaths <- catMaybes <$> mapM (normalizeStorePath store) sps +        -- Resolve paths and track which ones are invalid+        (errors, validPaths) <- liftIO $ resolveStorePaths store sps++        -- Log warnings for invalid paths+        liftIO $ for_ errors $ uncurry logStorePathWarning++        -- Emit PushStorePathInvalid events for invalid paths+        forM_ errors $ \(path, err) ->+          sendStorePathEvent [pushId] (PushStorePathInvalid path (formatStorePathError err))+         paths <- computeClosure store validPaths          -- Use async batch manager for narinfo queries (non-blocking)@@ -422,16 +452,25 @@        onUncompressedNARStream _ size = do         sp <- liftIO $ storePathToPath store storePath-        lastEmitRef <- liftIO $ newIORef (0 :: Int64)+        progressEmitIntervalNs <- asks pmProgressEmitIntervalNs+        lastEmitNsRef <- liftIO $ newIORef =<< getMonotonicTimeNSec         currentBytesRef <- liftIO $ newIORef (0 :: Int64)+        lastEmittedBytesRef <- liftIO $ newIORef (0 :: Int64)         C.awaitForever $ \chunk -> do           let newBytes = fromIntegral (BS.length chunk)           currentBytes <- liftIO $ atomicModifyIORef' currentBytesRef (\b -> (b + newBytes, b + newBytes))-          lastEmit <- liftIO $ readIORef lastEmitRef+          lastEmitNs <- liftIO $ readIORef lastEmitNsRef+          nowNs <- liftIO getMonotonicTimeNSec -          when (currentBytes - lastEmit >= 1024 || currentBytes == size) $ do-            liftIO $ writeIORef lastEmitRef currentBytes-            lift $ lift $ pushStorePathProgress (toS sp) currentBytes newBytes+          when (nowNs - lastEmitNs >= progressEmitIntervalNs || currentBytes == size) $ do+            liftIO $ writeIORef lastEmitNsRef nowNs+            lastEmitted <- liftIO $ readIORef lastEmittedBytesRef+            let emitBytes = currentBytes - lastEmitted+            liftIO $ writeIORef lastEmittedBytesRef currentBytes+            when (emitBytes > 0) $+              lift $+                lift $+                  pushStorePathProgress (toS sp) currentBytes emitBytes            C.yield chunk @@ -537,13 +576,6 @@   fp <- liftIO $ storePathToPath store storePath   pure $ toS fp --- | Canonicalize and validate a store path-normalizeStorePath :: (MonadIO m) => Store -> FilePath -> m (Maybe StorePath)-normalizeStorePath store fp =-  liftIO $ runMaybeT $ do-    storePath <- MaybeT $ followLinksToStorePath store (encodeUtf8 $ T.pack fp)-    MaybeT $ filterInvalidStorePath store storePath- withException :: (E.MonadCatch m) => m a -> (SomeException -> m a) -> m a withException action handler = action `E.catchAll` (\e -> handler e >> E.throwM e) @@ -564,21 +596,23 @@ decrementTVar tvar = modifyTVar' tvar (subtract 1)  -- | Run a transaction with a timeout.+-- Returns True if the transaction completed, False if timed out. atomicallyWithTimeout ::   TimeoutOptions ->   -- | A TVar timestamp to compare against   TVar UTCTime ->   -- | The transaction to run   STM () ->-  IO ()+  IO Bool atomicallyWithTimeout TimeoutOptions {..} timeVar transaction = do   timeoutVar <- newTVarIO False-  Async.race_-    (updateShutdownTimeout timeoutVar)-    (waitForGracefulShutdown timeoutVar)+  Async.withAsync (updateShutdownTimeout timeoutVar) $ \_ ->+    waitForGracefulShutdown timeoutVar   where     waitForGracefulShutdown timeout =-      atomically $ transaction `orElse` checkShutdownTimeout timeout+      atomically $+        (transaction >> return True)+          `orElse` (checkShutdownTimeout timeout >> return False)      updateShutdownTimeout timeoutVar =       forever $ do
src/Cachix/Daemon/SocketStore.hs view
@@ -12,6 +12,7 @@  import Cachix.Daemon.Types.PushEvent (PushRequestId) import Cachix.Daemon.Types.SocketStore (Socket (..), SocketId, SocketStore (..))+import Control.Concurrent.MVar import Control.Concurrent.STM.TVar import Control.Monad.IO.Unlift (MonadUnliftIO) import Data.ByteString.Lazy qualified as LBS@@ -31,6 +32,7 @@ addSocket :: (MonadUnliftIO m) => Network.Socket.Socket -> (SocketId -> Network.Socket.Socket -> m ()) -> SocketStore -> m () addSocket socket handler (SocketStore st) = do   socketId <- newSocketId+  sendLock <- liftIO $ newMVar ()   handlerThread <- Async.async (handler socketId socket)   publisherThreads <- liftIO $ newTVarIO HashMap.empty   liftIO $ atomically $ modifyTVar' st $ HashMap.insert socketId (Socket {..})@@ -73,5 +75,5 @@ sendAll socketId msg (SocketStore stvar) = do   mSocket <- liftIO $ readTVarIO stvar   case HashMap.lookup socketId mSocket of-    Just (Socket {socket}) -> liftIO $ Socket.LBS.sendAll socket msg+    Just (Socket {socket, sendLock}) -> liftIO $ withMVar sendLock $ \_ -> Socket.LBS.sendAll socket msg     Nothing -> return ()
src/Cachix/Daemon/Types/PushEvent.hs view
@@ -33,6 +33,8 @@   | PushStorePathProgress FilePath Int64 Int64   | PushStorePathDone FilePath   | PushStorePathFailed FilePath Text+  | PushStorePathSkipped FilePath+  | PushStorePathInvalid FilePath Text   | PushFinished   deriving stock (Eq, Generic, Show)   deriving anyclass (FromJSON, ToJSON)
src/Cachix/Daemon/Types/PushManager.hs view
@@ -21,6 +21,7 @@ import Cachix.Daemon.NarinfoQuery (NarinfoQueryManager) import Cachix.Daemon.NarinfoQuery qualified as NarinfoQuery import Cachix.Daemon.Protocol qualified as Protocol+import Cachix.Daemon.ShutdownLatch (ShutdownLatch) import Cachix.Daemon.Types.Log (Logger) import Cachix.Daemon.Types.PushEvent (PushEvent (..)) import Cachix.Daemon.Types.TaskQueue (TaskQueue)@@ -73,10 +74,14 @@     pmOnPushEvent :: OnPushEvent,     -- | The timestamp of the most recent event. This is used to track activity internally.     pmLastEventTimestamp :: TVar UTCTime,+    -- | Minimum interval between progress events (monotonic time, in nanoseconds).+    pmProgressEmitIntervalNs :: Word64,     -- | The number of pending (uncompleted) jobs.     pmPendingJobCount :: TVar Int,     -- | Manager for batching narinfo queries     pmNarinfoQueryManager :: NarinfoQueryManager Protocol.PushRequestId,+    -- | Latch to coordinate graceful shutdown of the push pipeline+    pmShutdownLatch :: ShutdownLatch () (),     pmLogger :: Logger   }
src/Cachix/Daemon/Types/SocketStore.hs view
@@ -10,6 +10,7 @@ data Socket = Socket   { socketId :: SocketId,     socket :: Network.Socket,+    sendLock :: MVar (),     handlerThread :: Async (),     publisherThreads :: TVar (HashMap PushRequestId (Async ()))   }
test/Daemon/NarinfoQuerySpec.hs view
@@ -80,17 +80,13 @@  -- Helper to start batch processor asynchronously with its own Katip context startQueryProcessorAsync :: NarinfoQueryManager requestId -> ([StorePath] -> IO ([StorePath], [StorePath])) -> IO ()-startQueryProcessorAsync manager batchProcessor = do-  started <- newEmptyMVar+startQueryProcessorAsync manager batchProcessor =   void $ Async.async $ do     handleScribe <- Katip.mkHandleScribe Katip.ColorIfTerminal stderr (Katip.permitItem Katip.InfoS) Katip.V0     let makeLogEnv = Katip.registerScribe "stderr" handleScribe Katip.defaultScribeSettings =<< Katip.initLogEnv "test" "test"     bracket makeLogEnv Katip.closeScribes $ \le ->-      Katip.runKatipContextT le () mempty $ do-        liftIO $ putMVar started ()+      Katip.runKatipContextT le () mempty $         NarinfoQuery.start manager (liftIO . batchProcessor)-  -- Wait for the processor to start before returning-  takeMVar started  -- Test setup helper that encapsulates common initialization withTestManager ::@@ -128,6 +124,13 @@   when (isNothing result) $     expectationFailure "Timeout waiting for STM condition" +-- | Wait for at least @n@ callbacks to be recorded, with a 5s timeout.+waitForCallbacks :: TVar [CallbackCall requestId] -> Int -> IO ()+waitForCallbacks callsVar n =+  waitForSTM 5_000_000 $ do+    cbs <- readTVar callsVar+    return $ length cbs >= n+ spec :: Spec spec = do   -- Initialize the CNix library@@ -142,27 +145,21 @@       withTestManager config $ \TestContext {..} -> do         atomically $ writeTVar tcResponsesQueue [(Set.fromList [path1, path2, path3], Set.empty)] -        -- Submit first request (1 path) - should not trigger+        -- Submit both requests - second one reaches size threshold (2 paths total)         NarinfoQuery.submitRequest tcManager (1 :: Int) [path1]-        threadDelay 10000+        NarinfoQuery.submitRequest tcManager (2 :: Int) [path2] -        calls1 <- readTVarIO tcBatchCalls-        length calls1 `shouldBe` 0 -- No batch yet+        -- Wait for the batch to be processed and both callbacks to arrive+        waitForCallbacks tcCallbackCalls 2 -        -- Submit second request (1 more unique path) - should trigger batch (2 paths total)-        NarinfoQuery.submitRequest tcManager (2 :: Int) [path2]-        threadDelay 20000 -- Wait for batch processing-        calls2 <- readTVarIO tcBatchCalls-        length calls2 `shouldBe` 1 -- One batch triggered-        let batchPaths = case head calls2 of+        -- Verify exactly one batch with both paths+        calls <- readTVarIO tcBatchCalls+        length calls `shouldBe` 1+        let batchPaths = case head calls of               Just (BatchCall paths _) -> paths               Nothing -> panic "Expected batch call"         Set.fromList batchPaths `shouldBe` Set.fromList [path1, path2] -        -- Verify both requests got responses-        callbacks <- readTVarIO tcCallbackCalls-        length callbacks `shouldBe` 2-     it "triggers batch when timeout is reached" $ withStoreFromURI "dummy://" $ \store -> do       path1 <- mockStorePath store 1       let config = defaultNarinfoQueryOptions {nqoMaxBatchSize = 100, nqoMaxWaitTime = 0.05} -- Small timeout, large batch@@ -172,12 +169,11 @@         -- Submit request that won't reach size threshold         NarinfoQuery.submitRequest tcManager (1 :: Int) [path1] -        -- Wait for timeout to trigger-        threadDelay 60000 -- 60ms > 50ms timeout+        -- Wait for timeout-triggered batch and callback+        waitForCallbacks tcCallbackCalls 1+         calls <- readTVarIO tcBatchCalls         length calls `shouldBe` 1 -- Batch triggered by timeout-        callbacks <- readTVarIO tcCallbackCalls-        length callbacks `shouldBe` 1 -- Request got response     it "processes immediately when timeout is zero" $ withStoreFromURI "dummy://" $ \store -> do       path1 <- mockStorePath store 1       let config = defaultNarinfoQueryOptions {nqoMaxBatchSize = 100, nqoMaxWaitTime = 0} -- Immediate mode@@ -185,12 +181,11 @@         atomically $ writeTVar tcResponsesQueue [(Set.fromList [path1], Set.empty)]          NarinfoQuery.submitRequest tcManager (1 :: Int) [path1]-        threadDelay 20000 -- Short wait++        waitForCallbacks tcCallbackCalls 1+         calls <- readTVarIO tcBatchCalls         length calls `shouldBe` 1 -- Processed immediately-        callbacks <- readTVarIO tcCallbackCalls-        length callbacks `shouldBe` 1-     it "only caches existing paths, not missing ones" $ withStoreFromURI "dummy://" $ \store -> do       path1 <- mockStorePath store 1       path2 <- mockStorePath store 2@@ -203,8 +198,9 @@          -- Submit all three paths         NarinfoQuery.submitRequest tcManager (1 :: Int) [path1, path2, path3]-        threadDelay 20000 +        waitForCallbacks tcCallbackCalls 1+         -- Check what's in cache - only existing paths should be cached         cached1 <- NarinfoQuery.lookupCache tcManager path1         cached2 <- NarinfoQuery.lookupCache tcManager path2@@ -222,12 +218,14 @@          -- First request - path1 will be cached         NarinfoQuery.submitRequest tcManager (1 :: Int) [path1]-        threadDelay 20000 +        waitForCallbacks tcCallbackCalls 1+         -- Second request - path1 from cache, path2 goes to batch         NarinfoQuery.submitRequest tcManager (2 :: Int) [path1, path2]-        threadDelay 20000 +        waitForCallbacks tcCallbackCalls 2+         -- Should have 2 batch calls (one for each unique uncached path)         calls <- readTVarIO tcBatchCalls         length calls `shouldBe` 2@@ -238,11 +236,8 @@               Nothing -> panic "Expected batch call"         secondBatchPaths `shouldBe` [path2] -        -- Both requests should have gotten responses-        callbacks <- readTVarIO tcCallbackCalls-        length callbacks `shouldBe` 2-         -- Second response should contain both paths (path1 from cache + path2 from batch)+        callbacks <- readTVarIO tcCallbackCalls         let secondResponse = case head callbacks of               Just (CallbackCall _ response _) -> response               Nothing -> panic "Expected callback call"@@ -270,12 +265,9 @@         NarinfoQuery.submitRequest tcManager (2 :: Int) [path3, path4, path5]          -- Wait until both callbacks are received (deterministic, no timing dependency)-        waitForSTM 5_000_000 $ do-          cbs <- readTVar tcCallbackCalls-          return $ length cbs >= 2+        waitForCallbacks tcCallbackCalls 2          callbacks <- readTVarIO tcCallbackCalls-        length callbacks `shouldBe` 2          -- Find responses by request ID         let findResponse rid = find (\(CallbackCall r _ _) -> r == rid) callbacks@@ -300,7 +292,8 @@         -- Submit overlapping requests that will be batched together         NarinfoQuery.submitRequest tcManager (1 :: Int) [path1, path2] -- paths 1,2         NarinfoQuery.submitRequest tcManager (2 :: Int) [path2, path1] -- paths 2,1 (same, different order)-        threadDelay 120000 -- Wait 120ms, longer than 100ms timeout+        waitForCallbacks tcCallbackCalls 2+         calls <- readTVarIO tcBatchCalls         length calls `shouldBe` 1 @@ -310,7 +303,3 @@               Nothing -> panic "Expected batch call"         Set.fromList batchPaths `shouldBe` Set.fromList [path1, path2]         length batchPaths `shouldBe` 2 -- No duplicates--        -- Both requests should still get responses-        callbacks <- readTVarIO tcCallbackCalls-        length callbacks `shouldBe` 2
test/Daemon/PostBuildHookSpec.hs view
@@ -4,12 +4,42 @@ import Data.String import Protolude import System.Environment qualified as System+import System.FilePath (takeBaseName, takeDirectory, (</>))+import System.IO.Temp (getCanonicalTemporaryDirectory) import Test.Hspec  spec :: Spec spec = do   let scriptPath = "build-hook.sh"       configPath = "my-nix.conf"++  describe "withRunnerFriendlyTempDirectory" $ do+    it "creates a temp directory with a socket path that fits within the limit" $ do+      let prefix = "cachix-daemon"+      withRunnerFriendlyTempDirectory prefix $ \tempDir -> do+        let dirName = takeBaseName tempDir+            socketPath = tempDir </> "daemon.sock"+        length dirName `shouldBe` length prefix + 6+        length socketPath `shouldSatisfy` (< unixSocketMaxPath)++    it "uses RUNNER_TEMP when path fits within socket limit" $ do+      systemTempDir <- getCanonicalTemporaryDirectory+      withTempEnv ("RUNNER_TEMP", systemTempDir) $ do+        withRunnerFriendlyTempDirectory "cachix-daemon" $ \tempDir -> do+          takeDirectory tempDir `shouldBe` systemTempDir++    it "falls back to system temp when RUNNER_TEMP is unset or too long" $ do+      systemTempDir <- getCanonicalTemporaryDirectory++      withTempEnv ("RUNNER_TEMP", "") $ do+        System.unsetEnv "RUNNER_TEMP"+        withRunnerFriendlyTempDirectory "cachix-daemon" $ \tempDir -> do+          takeDirectory tempDir `shouldBe` systemTempDir++      let longPath = "/tmp" </> replicate (unixSocketMaxPath - 20) 'x'+      withTempEnv ("RUNNER_TEMP", longPath) $ do+        withRunnerFriendlyTempDirectory "cachix-daemon" $ \tempDir -> do+          takeDirectory tempDir `shouldBe` systemTempDir    describe "post build hook" $ do     it "builds the NIX_CONF environment variable" $ do
test/Daemon/PushManagerSpec.hs view
@@ -149,8 +149,9 @@      describe "graceful shutdown" $ do       it "shuts down with no jobs" $-        withPushManager $-          stopPushManager timeoutOptions+        withPushManager $ \pm -> do+          _ <- drainPushManager timeoutOptions pm+          closePushManager pm        it "shuts down after jobs complete" $ withPushManager $ \pm -> do         let paths = ["foo"]@@ -165,13 +166,14 @@           return pushId          startTime <- getCurrentTime-        concurrently_ (stopPushManager longTimeoutOptions pm) $+        concurrently_ (drainPushManager longTimeoutOptions pm) $           runPushManager pm $             for_ paths pushStorePathDone         endTime <- getCurrentTime          let elapsed = diffUTCTime endTime startTime         elapsed `shouldSatisfy` (< 0.5)+        closePushManager pm        it "shuts down on job stall" $         withPushManager $ \pm -> do@@ -179,13 +181,15 @@             let request = Protocol.PushRequest {Protocol.storePaths = ["foo"], Protocol.subscribeToUpdates = False}             addPushJobFromRequest request -          stopPushManager timeoutOptions pm+          _ <- drainPushManager timeoutOptions pm+          closePushManager pm    describe "STM" $     describe "timeout" $ do       it "times out a transaction after n seconds" $ do         timestamp <- newTVarIO =<< getCurrentTime-        atomicallyWithTimeout timeoutOptions timestamp retry+        result <- atomicallyWithTimeout timeoutOptions timestamp retry+        result `shouldBe` False  withPushManager :: (PushManagerEnv -> IO a) -> IO a withPushManager f = do