diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 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.10.0 +version: 1.11.0 synopsis: Command-line client for Nix binary cache hosting https://cachix.org diff --git a/src/Cachix/Client/CNix.hs b/src/Cachix/Client/CNix.hs --- a/src/Cachix/Client/CNix.hs +++ b/src/Cachix/Client/CNix.hs @@ -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. -- diff --git a/src/Cachix/Client/Command/Import.hs b/src/Cachix/Client/Command/Import.hs --- a/src/Cachix/Client/Command/Import.hs +++ b/src/Cachix/Client/Command/Import.hs @@ -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 -} diff --git a/src/Cachix/Client/Command/Pin.hs b/src/Cachix/Client/Command/Pin.hs --- a/src/Cachix/Client/Command/Pin.hs +++ b/src/Cachix/Client/Command/Pin.hs @@ -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 diff --git a/src/Cachix/Client/Command/Push.hs b/src/Cachix/Client/Command/Push.hs --- a/src/Cachix/Client/Command/Push.hs +++ b/src/Cachix/Client/Command/Push.hs @@ -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." diff --git a/src/Cachix/Client/Command/Watch.hs b/src/Cachix/Client/Command/Watch.hs --- a/src/Cachix/Client/Command/Watch.hs +++ b/src/Cachix/Client/Command/Watch.hs @@ -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 diff --git a/src/Cachix/Client/Exception.hs b/src/Cachix/Client/Exception.hs --- a/src/Cachix/Client/Exception.hs +++ b/src/Cachix/Client/Exception.hs @@ -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 diff --git a/src/Cachix/Client/OptionsParser.hs b/src/Cachix/Client/OptionsParser.hs --- a/src/Cachix/Client/OptionsParser.hs +++ b/src/Cachix/Client/OptionsParser.hs @@ -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 = diff --git a/src/Cachix/Client/Push.hs b/src/Cachix/Client/Push.hs --- a/src/Cachix/Client/Push.hs +++ b/src/Cachix/Client/Push.hs @@ -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 diff --git a/src/Cachix/Client/Push/S3.hs b/src/Cachix/Client/Push/S3.hs --- a/src/Cachix/Client/Push/S3.hs +++ b/src/Cachix/Client/Push/S3.hs @@ -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 -} diff --git a/src/Cachix/Client/PushQueue.hs b/src/Cachix/Client/PushQueue.hs --- a/src/Cachix/Client/PushQueue.hs +++ b/src/Cachix/Client/PushQueue.hs @@ -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 diff --git a/src/Cachix/Client/WatchStore.hs b/src/Cachix/Client/WatchStore.hs --- a/src/Cachix/Client/WatchStore.hs +++ b/src/Cachix/Client/WatchStore.hs @@ -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] diff --git a/src/Cachix/Daemon.hs b/src/Cachix/Daemon.hs --- a/src/Cachix/Daemon.hs +++ b/src/Cachix/Daemon.hs @@ -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 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 @@ -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 diff --git a/src/Cachix/Daemon/Listen.hs b/src/Cachix/Daemon/Listen.hs --- a/src/Cachix/Daemon/Listen.hs +++ b/src/Cachix/Daemon/Listen.hs @@ -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 diff --git a/src/Cachix/Daemon/NarinfoQuery.hs b/src/Cachix/Daemon/NarinfoQuery.hs --- a/src/Cachix/Daemon/NarinfoQuery.hs +++ b/src/Cachix/Daemon/NarinfoQuery.hs @@ -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 diff --git a/src/Cachix/Daemon/PostBuildHook.hs b/src/Cachix/Daemon/PostBuildHook.hs --- a/src/Cachix/Daemon/PostBuildHook.hs +++ b/src/Cachix/Daemon/PostBuildHook.hs @@ -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 ()) diff --git a/src/Cachix/Daemon/Progress.hs b/src/Cachix/Daemon/Progress.hs --- a/src/Cachix/Daemon/Progress.hs +++ b/src/Cachix/Daemon/Progress.hs @@ -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 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 @@ -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 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 @@ -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 () diff --git a/src/Cachix/Daemon/Types/PushEvent.hs b/src/Cachix/Daemon/Types/PushEvent.hs --- a/src/Cachix/Daemon/Types/PushEvent.hs +++ b/src/Cachix/Daemon/Types/PushEvent.hs @@ -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) diff --git a/src/Cachix/Daemon/Types/PushManager.hs b/src/Cachix/Daemon/Types/PushManager.hs --- a/src/Cachix/Daemon/Types/PushManager.hs +++ b/src/Cachix/Daemon/Types/PushManager.hs @@ -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 } 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 @@ -10,6 +10,7 @@ data Socket = Socket { socketId :: SocketId, socket :: Network.Socket, + sendLock :: MVar (), handlerThread :: Async (), publisherThreads :: TVar (HashMap PushRequestId (Async ())) } diff --git a/test/Daemon/NarinfoQuerySpec.hs b/test/Daemon/NarinfoQuerySpec.hs --- a/test/Daemon/NarinfoQuerySpec.hs +++ b/test/Daemon/NarinfoQuerySpec.hs @@ -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 diff --git a/test/Daemon/PostBuildHookSpec.hs b/test/Daemon/PostBuildHookSpec.hs --- a/test/Daemon/PostBuildHookSpec.hs +++ b/test/Daemon/PostBuildHookSpec.hs @@ -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 diff --git a/test/Daemon/PushManagerSpec.hs b/test/Daemon/PushManagerSpec.hs --- a/test/Daemon/PushManagerSpec.hs +++ b/test/Daemon/PushManagerSpec.hs @@ -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