diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,26 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.9.4] - 2022-05-17 + +### Added + + - Concurrent IFD, reducing evaluation wall clock time + - Nix 2.8.0 support + - Improved log contexts + +### Fixed + + - Workaround for cachix#406 (add `login` to `netrc`) + - A crash in `inline-c-cpp` exception handling (`inline-c-cpp` update) + - Towards the error "Could not push logs within 10 minutes after completion" + - Add a timeout to prevent in case of a stuck handshake + - Enforce log limit on client side as well in case of excessive log spam and an upload bottleneck + +### Removed + + - `hercules-ci-agent-nix_2_5` variant: upgrade to plain `hercules-ci-agent` (2.8.0) or `_nix_2_7`. + ## [0.9.3] - 2022-04-08 ### Added @@ -563,6 +583,7 @@ - Initial release +[0.9.3]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.2...hercules-ci-agent-0.9.3 [0.9.2]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.1...hercules-ci-agent-0.9.2 [0.9.1]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.0...hercules-ci-agent-0.9.1 [0.9.0]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.8.7...hercules-ci-agent-0.9.0 diff --git a/cbits/hercules-error.cxx b/cbits/hercules-error.cxx new file mode 100644 --- /dev/null +++ b/cbits/hercules-error.cxx @@ -0,0 +1,29 @@ +#include "hercules-error.hh" + +void +hercules::copyErrorStrings(const nix::Error &err, const char **msgStrPtr, const char **traceStrPtr) noexcept { + std::string msg; + { + std::stringstream s; + nix::showErrorInfo(s, err.info(), false); + msg = s.str(); + *msgStrPtr = strdup(msg.c_str()); + } + + // There's no method for getting just the trace, short of + // reinventing the printing ourselves. That didn't seem to + // be worth the effort. + { + std::stringstream s; + nix::showErrorInfo(s, err.info(), true); + std::string t = s.str(); + + // drop prefix msg + if (t.rfind(msg, 0) == 0) { + t.replace(0, msg.size(), ""); + t = nix::trim(t); + } + + *traceStrPtr = strdup(t.c_str()); + } +} diff --git a/cbits/nix-2.4/hercules-store.cxx b/cbits/nix-2.4/hercules-store.cxx --- a/cbits/nix-2.4/hercules-store.cxx +++ b/cbits/nix-2.4/hercules-store.cxx @@ -216,14 +216,16 @@ downloadSize, narSize); } -#if NIX_IS_AT_LEAST(2,6,0) +#if !NIX_IS_AT_LEAST(2,8,0) +# if NIX_IS_AT_LEAST(2,6,0) std::optional -#else +# else std::shared_ptr -#endif +# endif WrappingStore::getBuildLog(const StorePath& path) { return wrappedStore->getBuildLog(path); } +#endif void WrappingStore::connect() { wrappedStore->connect(); diff --git a/cbits/nix-2.4/hercules-store.hh b/cbits/nix-2.4/hercules-store.hh --- a/cbits/nix-2.4/hercules-store.hh +++ b/cbits/nix-2.4/hercules-store.hh @@ -135,12 +135,14 @@ StorePathSet & willBuild, StorePathSet & willSubstitute, StorePathSet & unknown, uint64_t & downloadSize, uint64_t & narSize) override; -#if NIX_IS_AT_LEAST(2,6,0) +#if !NIX_IS_AT_LEAST(2,8,0) +# if NIX_IS_AT_LEAST(2,6,0) virtual std::optional -#else +# else virtual std::shared_ptr -#endif +# endif getBuildLog(const StorePath & path) override; +#endif virtual unsigned int getProtocol() override; diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker.hs --- a/hercules-ci-agent-worker/Hercules/Agent/Worker.hs +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker.hs @@ -2,10 +2,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NumericUnderscores #-} -{-# LANGUAGE OverloadedLabels #-} -{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TemplateHaskell #-} module Hercules.Agent.Worker ( main, @@ -18,8 +15,6 @@ import Control.Monad.Except import Control.Monad.IO.Unlift import Control.Monad.Trans.Control -import Data.ByteString.Unsafe (unsafePackMallocCString) -import Data.Coerce (coerce) import qualified Data.Conduit import Data.Conduit.Extras (sinkChan, sinkChanTerminate, sourceChan) import Data.Conduit.Katip.Orphans () @@ -29,34 +24,24 @@ ) import Data.IORef import qualified Data.Map as M -import qualified Data.Set as S -import Data.UUID (UUID) import Data.Vector (Vector) import qualified Data.Vector as V -import Foreign (alloca, peek) -import Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent (OnPushHandlerEvent (OnPushHandlerEvent)) -import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent -import qualified Hercules.API.Agent.Evaluate.EvaluateTask as EvaluateTask -import qualified Hercules.API.Agent.Evaluate.EvaluateTask.OnPush as OnPush -import qualified Hercules.API.Agent.Evaluate.ImmutableGitInput as API.ImmutableGitInput -import qualified Hercules.API.Agent.Evaluate.ImmutableInput as API.ImmutableInput import qualified Hercules.API.Agent.LifeCycle.ServiceInfo import Hercules.API.Logs.LogEntry (LogEntry) import qualified Hercules.API.Logs.LogEntry as LogEntry import Hercules.API.Logs.LogHello (LogHello (LogHello, clientProtocolVersion, storeProtocolVersion)) import Hercules.API.Logs.LogMessage (LogMessage) import qualified Hercules.API.Logs.LogMessage as LogMessage -import Hercules.Agent.NixFile (HerculesCISchema, getHerculesCI, homeExprRawValue, loadNixFile, parseExtraInputs) -import qualified Hercules.Agent.NixFile as NixFile -import Hercules.Agent.NixFile.HerculesCIArgs (CISystems (CISystems), HerculesCIMeta (HerculesCIMeta), fromGitSource) -import qualified Hercules.Agent.NixFile.HerculesCIArgs import Hercules.Agent.Sensitive import qualified Hercules.Agent.Socket as Socket import Hercules.Agent.Worker.Build (runBuild) import qualified Hercules.Agent.Worker.Build.Logger as Logger +import Hercules.Agent.Worker.Conduit (takeCWhileStopEarly, withMessageLimit) import Hercules.Agent.Worker.Effect (runEffect) -import Hercules.Agent.Worker.HerculesStore (nixStore, setBuilderCallback, withHerculesStore) -import Hercules.Agent.Worker.HerculesStore.Context (HerculesStore) +import Hercules.Agent.Worker.Env (HerculesState (..)) +import Hercules.Agent.Worker.Error (ExceptionText (exceptionTextMessage), exceptionTextMessage, renderException) +import Hercules.Agent.Worker.Evaluate (runEval) +import Hercules.Agent.Worker.HerculesStore (setBuilderCallback, withHerculesStore) import Hercules.Agent.Worker.Logging (withKatip) import Hercules.Agent.WorkerProtocol.Command ( Command, @@ -71,65 +56,25 @@ import qualified Hercules.Agent.WorkerProtocol.Command.Eval as Eval import Hercules.Agent.WorkerProtocol.Event ( Event (Exception), - ViaJSON (ViaJSON), ) import qualified Hercules.Agent.WorkerProtocol.Event as Event -import qualified Hercules.Agent.WorkerProtocol.Event.Attribute as Attribute -import qualified Hercules.Agent.WorkerProtocol.Event.AttributeError as AttributeError import qualified Hercules.Agent.WorkerProtocol.LogSettings as LogSettings import Hercules.CNix as CNix -import Hercules.CNix.Expr (Match (IsAttrs, IsString), NixAttrs, RawValue, addAllowedPath, addInternalAllowedPaths, autoCallFunction, evalArgs, getAttrBool, getAttrList, getAttrs, getDrvFile, getFlakeFromArchiveUrl, getFlakeFromGit, getRecurseForDerivations, getStringIgnoreContext, init, isDerivation, isFunctor, match, rawValueType, rtValue, toRawValue, toValue, withEvalStateConduit) -import Hercules.CNix.Expr.Context (EvalState) -import qualified Hercules.CNix.Expr.Raw -import Hercules.CNix.Expr.Schema (MonadEval, PSObject, dictionaryToMap, fromPSObject, requireDict, (#.), (#?), (#?!), ($?)) -import qualified Hercules.CNix.Expr.Schema as Schema -import Hercules.CNix.Expr.Typed (Value) -import Hercules.CNix.Std.Vector (StdVector) -import qualified Hercules.CNix.Std.Vector as Std.Vector -import Hercules.CNix.Store.Context (NixStorePathWithOutputs) +import Hercules.CNix.Expr (init) import Hercules.CNix.Util (installDefaultSigINTHandler) import Hercules.CNix.Verbosity (setShowTrace) import Hercules.Error -import Hercules.UserException (UserException (UserException)) import Katip -import qualified Language.C.Inline.Cpp as C -import qualified Language.C.Inline.Cpp.Exception as C import qualified Network.URI import Protolude hiding (bracket, catch, check, evalState, wait, withAsync, yield) import qualified System.Environment as Environment import System.IO (BufferMode (LineBuffering), hSetBuffering) import System.Posix.Signals (Handler (Catch), installHandler, raiseSignal, sigINT, sigTERM) import System.Timeout (timeout) -import qualified UnliftIO import UnliftIO.Async (wait, withAsync) -import UnliftIO.Exception (bracket, catch) import Prelude () import qualified Prelude -C.context (C.cppCtx <> C.fptrCtx) - -C.include "" -C.include "" -C.include "" -C.include "" - -data HerculesState = HerculesState - { drvsCompleted :: TVar (Map StorePath (UUID, BuildResult.BuildStatus)), - drvsInProgress :: IORef (Set StorePath), - herculesStore :: Ptr (Ref HerculesStore), - wrappedStore :: Store, - shortcutChannel :: Chan (Maybe Event), - extraNixOptions :: [(Text, Text)] - } - -data BuildException = BuildException - { buildExceptionDerivationPath :: Text, - buildExceptionDetail :: Maybe Text - } - deriving (Show, Typeable) - -instance Exception BuildException - main :: IO () main = do hSetBuffering stderr LineBuffering @@ -165,6 +110,9 @@ taskWorker options = do setOptions options drvsCompleted_ <- newTVarIO mempty + drvBuildAsyncs_ <- newTVarIO mempty + drvRebuildAsyncs_ <- newTVarIO mempty + drvOutputSubstituteAsyncs_ <- newTVarIO mempty drvsInProgress_ <- newIORef mempty withStore $ \wrappedStore_ -> withHerculesStore wrappedStore_ $ \herculesStore_ -> withKatip $ do liftIO $ setBuilderCallback herculesStore_ mempty @@ -172,6 +120,9 @@ let st = HerculesState { drvsCompleted = drvsCompleted_, + drvBuildAsyncs = drvBuildAsyncs_, + drvRebuildAsyncs = drvRebuildAsyncs_, + drvOutputSubstituteAsyncs = drvOutputSubstituteAsyncs_, drvsInProgress = drvsInProgress_, herculesStore = herculesStore_, wrappedStore = wrappedStore_, @@ -194,8 +145,8 @@ runCommand st ch command ) `safeLiftedCatch` ( \e -> liftIO $ do - (e', _trace) <- renderException e - writeChan ch (Just $ Exception e') + textual <- renderException e + writeChan ch (Just $ Exception $ exceptionTextMessage textual) exitFailure ) ) @@ -225,57 +176,6 @@ pure x ) -renderException :: SomeException -> IO (Text, Maybe Text) -renderException e | Just (C.CppStdException ex _msg _ty) <- fromException e = renderStdException ex -renderException e - | Just (C.CppNonStdException _ex maybeType) <- fromException e = - pure ("Unexpected C++ exception" <> foldMap (\t -> " of type " <> decodeUtf8With lenientDecode t) maybeType, Nothing) -renderException e | Just (FatalError msg) <- fromException e = pure (msg, Nothing) -renderException e = pure (toS $ displayException e, Nothing) - -renderStdException :: C.CppExceptionPtr -> IO (Text, Maybe Text) -renderStdException e = alloca \traceStrPtr -> do - msg <- - [C.throwBlock| char * { - char **traceStrPtr = $(char **traceStrPtr); - *traceStrPtr = nullptr; - std::string r; - std::exception_ptr *e = $fptr-ptr:(std::exception_ptr *e); - try { - std::rethrow_exception(*e); - } catch (const nix::Error &e) { - { - std::stringstream s; - nix::showErrorInfo(s, e.info(), false); - r = s.str(); - } - { - std::stringstream s; - nix::showErrorInfo(s, e.info(), true); - std::string t = s.str(); - // starts with r? - if (t.rfind(r, 0) == 0) { - t.replace(0, r.size(), ""); - t = nix::trim(t); - } - *traceStrPtr = strdup(t.c_str()); - } - } catch (const std::exception &e) { - r = e.what(); - } catch (...) { - // shouldn't happen because inline-c-cpp only put std::exception in CppStdException - throw std::runtime_error("renderStdException: Attempt to render unknown exception."); - } - - return strdup(r.c_str()); - }|] - >>= unsafePackMallocCString - <&> decodeUtf8With lenientDecode - traceText <- - peek traceStrPtr >>= traverseNonNull \s -> - unsafePackMallocCString s <&> decodeUtf8With lenientDecode - pure (msg, traceText) - connectCommand :: (MonadUnliftIO m, KatipContext m, MonadThrow m) => Chan (Maybe Event) -> @@ -312,7 +212,7 @@ runConduitRes ( Data.Conduit.handleC ( \e -> do - yield . Event.Error . fst =<< liftIO (renderException e) + yield . Event.Error . exceptionTextMessage =<< liftIO (renderException e) liftIO $ throwTo mainThread e ) ( do @@ -374,6 +274,7 @@ entriesSource .| Logger.unbatch .| Logger.filterProgress + .| dropMiddle .| renumber 0 .| batchAndEnd .| socketSink socket @@ -398,6 +299,67 @@ Nothing -> panic "Could not push logs within 10 minutes after completion" logLocM DebugS "Logger done" +dropMiddle :: MonadIO m => ConduitM (Flush LogEntry) (Flush LogEntry) m () +dropMiddle = do + -- rich logging + takeCWhileStopEarly isChunk richLogLimit + -- degrade to text logging (in case rich logging produces excessive non-textual data) + visibleLinesOnly .| withMessageLimit isChunk textOnlyLogLimit tailLimit snipStart snip snipped + +-- Sum must be < 100_000 +richLogLimit, textOnlyLogLimit, tailLimit :: Int +richLogLimit = 40_000 +textOnlyLogLimit = 49_900 +tailLimit = 10_000 + +snipStart :: Monad m => ConduitT (Flush LogEntry) (Flush LogEntry) m () +snipStart = + yield $ + Chunk $ + LogEntry.Msg + { i = 0, + ms = 0, + level = 0 {- error -}, + msg = "hercules-ci-agent: Soft log limit has been reached. Final log lines will appear when done." + } + +snipped :: Monad m => Int -> ConduitT (Flush LogEntry) (Flush LogEntry) m () +snipped n = + yield $ + Chunk $ + LogEntry.Msg + { i = 0, + ms = 0, + level = 0 {- error -}, + msg = "hercules-ci-agent: " <> show n <> " log lines were omitted before the last " <> show tailLimit <> "." + } + +snip :: Monad m => Int -> ConduitT (Flush LogEntry) (Flush LogEntry) m () +snip n = + yield $ + Chunk $ + LogEntry.Msg + { i = 0, + ms = 0, + level = 0 {- error -}, + msg = "hercules-ci-agent: skipping " <> show n <> " log lines." + } + +visibleLinesOnly :: Monad m => ConduitM (Flush LogEntry) (Flush LogEntry) m () +visibleLinesOnly = + filterC isVisible + +isVisible :: Flush LogEntry -> Bool +isVisible Flush = True +isVisible (Chunk LogEntry.Msg {msg = msg}) | msg /= "" = True +isVisible (Chunk LogEntry.Start {text = msg}) | msg /= "" = True +isVisible (Chunk LogEntry.Result {rtype = LogEntry.ResultTypeBuildLogLine}) = True +isVisible _ = False + +isChunk :: Flush LogEntry -> Bool +isChunk Chunk {} = True +isChunk _ = False + socketSink :: MonadIO m => Socket.Socket r w -> ConduitT w o m () socketSink socket = awaitForever $ liftIO . atomically . Socket.write socket @@ -434,460 +396,3 @@ path = LogSettings.path l, token = encodeUtf8 $ reveal $ LogSettings.token l } - --- TODO: test -autoArgArgs :: Map Text Eval.Arg -> [ByteString] -autoArgArgs kvs = do - (k, v) <- M.toList kvs - case v of - Eval.LiteralArg s -> ["--argstr", encodeUtf8 k, s] - Eval.ExprArg s -> ["--arg", encodeUtf8 k, s] - --- Ensure that a nested build invocation does not happen by a mistake in wiring. --- --- (It does not happen, but this de-escalates a potential recursion bug to just --- an error.) -withDrvInProgress :: MonadUnliftIO m => HerculesState -> StorePath -> m a -> m a -withDrvInProgress HerculesState {drvsInProgress = ref} drvPath = - bracket acquire release . const - where - acquire = - liftIO $ - join $ - atomicModifyIORef ref $ \inprg -> - if drvPath `S.member` inprg - then (inprg, throwIO $ FatalError "Refusing to build derivation that should have been built remotely. Presumably, substitution has failed.") - else (S.insert drvPath inprg, pass) - release _ = - liftIO $ - atomicModifyIORef ref $ \inprg -> - (S.delete drvPath inprg, ()) - -anyAlternative :: (Foldable l, Alternative f) => l a -> f a -anyAlternative = getAlt . foldMap (Alt . pure) - -yieldAttributeError :: MonadIO m => [ByteString] -> SomeException -> ConduitT i Event m () -yieldAttributeError path e - | (Just e') <- fromException e = - yield $ - Event.AttributeError $ - AttributeError.AttributeError - { AttributeError.path = path, - AttributeError.message = - "Could not build derivation " <> buildExceptionDerivationPath e' - <> ", which is required during evaluation." - <> foldMap (" " <>) (buildExceptionDetail e'), - AttributeError.errorDerivation = Just (buildExceptionDerivationPath e'), - AttributeError.errorType = Just "BuildException", - AttributeError.trace = Nothing -- would be nice to get a trace here. Throw and catch more C++ exception types? - } -yieldAttributeError path e = do - (e', maybeTrace) <- liftIO $ renderException e - yield $ - Event.AttributeError $ - AttributeError.AttributeError - { AttributeError.path = path, - AttributeError.message = e', - AttributeError.errorDerivation = Nothing, - AttributeError.errorType = Just (show (typeOf e)), - AttributeError.trace = maybeTrace - } - -maybeThrowBuildException :: MonadIO m => BuildResult.BuildStatus -> Text -> m () -maybeThrowBuildException result plainDrvText = - case result of - BuildResult.Failure -> throwIO $ BuildException plainDrvText Nothing - BuildResult.DependencyFailure -> throwIO $ BuildException plainDrvText (Just "A dependency could not be built.") - BuildResult.Success -> pass - -mkCache :: forall k a m. (MonadUnliftIO m, Ord k) => IO (k -> m a -> m a) -mkCache = do - doneRef <- liftIO $ newIORef mempty - let hasBeenBuilt :: k -> m (Maybe (Either SomeException a)) - hasBeenBuilt p = liftIO $ readIORef doneRef <&> \c -> M.lookup p c - cacheBy p io = do - b <- hasBeenBuilt p - case b of - Just x -> liftIO $ escalate x - Nothing -> do - r <- UnliftIO.tryAny io - liftIO do - modifyIORef doneRef (M.insert p (r :: Either SomeException a)) - escalate r - pure cacheBy - -runEval :: - forall i m. - (MonadResource m, KatipContext m, MonadUnliftIO m, MonadThrow m) => - HerculesState -> - Eval -> - ConduitM i Event m () -runEval st@HerculesState {herculesStore = hStore, shortcutChannel = shortcutChan, drvsCompleted = drvsCompl} eval = do - for_ (Eval.extraNixOptions eval) $ liftIO . uncurry setGlobalOption - for_ (Eval.extraNixOptions eval) $ liftIO . uncurry setOption - let store = nixStore hStore - isFlake = Eval.isFlakeJob eval - s <- storeUri store - UnliftIO unlift <- lift askUnliftIO - let decode = decodeUtf8With lenientDecode - - cachingBuilt <- liftIO mkCache - - liftIO . setBuilderCallback hStore $ - traverseSPWOs $ \storePathWithOutputs -> unlift $ do - drvStorePath <- liftIO $ getStorePath storePathWithOutputs - drvPath <- liftIO $ CNix.storePathToPath store drvStorePath - cachingBuilt drvPath do - let pathText = decode drvPath - outputs <- liftIO $ getOutputs storePathWithOutputs - katipAddContext (sl "fullpath" pathText) $ - for_ outputs $ \outputName -> do - withDrvInProgress st drvStorePath $ do - derivation <- liftIO $ getDerivation store drvStorePath - drvName <- liftIO $ getDerivationNameFromPath drvStorePath - drvOutputs <- liftIO $ getDerivationOutputs store drvName derivation - outputPath <- - case find (\o -> derivationOutputName o == outputName) drvOutputs of - Nothing -> panic $ "output " <> show outputName <> " does not exist on " <> pathText - Just o -> case derivationOutputPath o of - Just x -> pure x - Nothing -> - -- FIXME ca-derivations - panic $ "output path unknown for output " <> show outputName <> " on " <> pathText <> ". ca-derivations is not supported yet." - - isValid <- liftIO $ isValidPath store outputPath - if isValid - then do - logLocM DebugS "Output already valid" - -- Report IFD - liftIO $ writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) Nothing False - else do - logLocM DebugS "Building" - liftIO $ writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) Nothing True - ( katipAddContext (sl "outputPath" (show outputPath :: Text)) do - logLocM DebugS "Attempting early ensurePath" - liftIO (ensurePath (wrappedStore st) outputPath) - ) - `catch` \e0 -> do - katipAddContext (sl "message" (show (e0 :: SomeException) :: Text)) $ - logLocM DebugS "Recovering from failed wrapped.ensurePath" - (attempt0, result) <- - liftIO $ - atomically $ do - c <- readTVar drvsCompl - anyAlternative $ M.lookup drvStorePath c - liftIO $ maybeThrowBuildException result (decode drvPath) - liftIO clearSubstituterCaches - liftIO $ clearPathInfoCache store - liftIO (ensurePath (wrappedStore st) outputPath) `catch` \e1 -> do - katipAddContext (sl "message" (show (e1 :: SomeException) :: Text)) $ - logLocM DebugS "Recovering from fresh ensurePath" - liftIO $ writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) (Just attempt0) True - -- TODO sync - result' <- - liftIO $ - atomically $ do - c <- readTVar drvsCompl - (attempt1, r) <- anyAlternative $ M.lookup drvStorePath c - guard (attempt1 /= attempt0) - pure r - liftIO $ maybeThrowBuildException result' (decode drvPath) - liftIO clearSubstituterCaches - liftIO $ clearPathInfoCache store - liftIO (ensurePath (wrappedStore st) outputPath) `catch` \e2 -> - liftIO $ - throwIO $ - BuildException - (decode drvPath) - ( Just $ - "It could not be retrieved on the evaluating agent, despite a successful rebuild. Exception: " - <> show (e2 :: SomeException) - ) - liftIO $ addTemporaryRoot store outputPath - logLocM DebugS "Built" - withEvalStateConduit store $ \evalState -> do - liftIO do - addInternalAllowedPaths evalState - for_ (Eval.allowedPaths eval) (addAllowedPath evalState) - katipAddContext (sl "storeURI" (decode s)) $ - logLocM DebugS "EvalState loaded." - args <- - liftIO $ - evalArgs evalState (autoArgArgs (Eval.autoArguments eval)) - Data.Conduit.handleC (yieldAttributeError []) $ - do - homeExpr <- getHomeExpr evalState eval - let hargs = fromGitSource (coerce $ Eval.gitSource eval) meta - meta = HerculesCIMeta {apiBaseUrl = Eval.apiBaseUrl eval, ciSystems = CISystems (Eval.ciSystems eval)} - liftIO (flip runReaderT evalState $ getHerculesCI homeExpr hargs) >>= \case - Nothing -> - -- legacy - walk store evalState args (homeExprRawValue homeExpr) - Just herculesCI -> do - case Event.fromViaJSON (Eval.selector eval) of - EvaluateTask.ConfigOrLegacy -> do - yield Event.JobConfig - sendConfig evalState isFlake herculesCI - EvaluateTask.OnPush onPush -> - transPipe (`runReaderT` evalState) do - walkOnPush store evalState onPush herculesCI - yield Event.EvaluationDone - -getHomeExpr :: (MonadThrow m, MonadIO m) => Ptr EvalState -> Eval -> m NixFile.HomeExpr -getHomeExpr evalState eval = - if Eval.isFlakeJob eval - then - NixFile.Flake <$> liftIO do - srcInput <- case Eval.srcInput eval of - Just x -> pure x - Nothing -> panic "srcInput is required for flake job" - raw <- mkImmutableGitInputFlakeThunk evalState (Event.fromViaJSON srcInput) - let pso :: PSObject (Schema.Attrs '[]) - pso = Schema.PSObject {value = raw, provenance = Schema.Other "flake.nix"} - toValue evalState pso - else escalateAs UserException =<< liftIO (loadNixFile evalState (toS $ Eval.cwd eval) (coerce $ Eval.gitSource eval)) - -walkOnPush :: (MonadEval m, MonadUnliftIO m, KatipContext m, MonadThrow m) => Store -> Ptr EvalState -> OnPush.OnPush -> PSObject HerculesCISchema -> ConduitT i Event m () -walkOnPush store evalState onPushParams herculesCI = do - onPushHandler <- herculesCI #?! #onPush >>= requireDict (OnPush.name onPushParams) - inputs <- liftIO $ do - inputs <- for (OnPush.inputs onPushParams) \input -> do - inputToValue evalState input - toRawValue evalState inputs - outputsFun <- onPushHandler #. #outputs - outputs <- outputsFun $? (Schema.PSObject {value = inputs, provenance = Schema.Data}) - simpleWalk store evalState (Schema.value outputs) - -inputToValue :: Ptr EvalState -> API.ImmutableInput.ImmutableInput -> IO RawValue -inputToValue evalState (API.ImmutableInput.ArchiveUrl u) = getFlakeFromArchiveUrl evalState u -inputToValue evalState (API.ImmutableInput.Git g) = mkImmutableGitInputFlakeThunk evalState g - -mkImmutableGitInputFlakeThunk :: Ptr EvalState -> API.ImmutableGitInput.ImmutableGitInput -> IO RawValue -mkImmutableGitInputFlakeThunk evalState git = do - -- TODO: allow picking ssh/http url - getFlakeFromGit - evalState - (API.ImmutableGitInput.httpURL git) - (API.ImmutableGitInput.ref git) - (API.ImmutableGitInput.rev git) - -sendConfig :: MonadIO m => Ptr EvalState -> Bool -> PSObject HerculesCISchema -> ConduitT i Event m () -sendConfig evalState isFlake herculesCI = flip runReaderT evalState $ do - herculesCI #? #onPush >>= traverse_ \onPushes -> do - attrs <- dictionaryToMap onPushes - for_ (M.mapWithKey (,) attrs) \(name, onPush) -> do - ei <- onPush #? #extraInputs >>= traverse parseExtraInputs - enable <- onPush #? #enable >>= traverse fromPSObject <&> fromMaybe True - when enable . lift . yield . Event.OnPushHandler . ViaJSON $ - OnPushHandlerEvent - { handlerName = decodeUtf8 name, - handlerExtraInputs = M.mapKeys decodeUtf8 (fromMaybe mempty ei), - isFlake = isFlake - } - --- | Documented in @docs/modules/ROOT/pages/evaluation.adoc@. -simpleWalk :: - (MonadUnliftIO m, KatipContext m) => - Store -> - Ptr EvalState -> - RawValue -> - ConduitT i Event m () -simpleWalk store evalState = walk' [] 10 - where - handleErrors path = Data.Conduit.handleC (yieldAttributeError path) - walk' :: - (MonadUnliftIO m, KatipContext m) => - -- Attribute path - [ByteString] -> - -- Depth of tree remaining - Integer -> - -- Current node of the walk - RawValue -> - -- Program that performs the walk and emits 'Event's - ConduitT i1 Event m () - walk' path depthRemaining v = - handleErrors path $ - liftIO (match evalState v) - >>= \case - Left e -> - yieldAttributeError path e - Right m -> case m of - IsAttrs attrValue -> do - isDeriv <- liftIO $ isDerivation evalState v - if isDeriv - then walkDerivation store evalState False path attrValue - else do - attrs <- liftIO $ getAttrs attrValue - void $ - flip M.traverseWithKey attrs $ - \name value -> - if depthRemaining > 0 - then - walk' - (path ++ [name]) - (depthRemaining - 1) - value - else yield (Event.Error $ "Max recursion depth reached at path " <> show path) - _any -> do - vt <- liftIO $ rawValueType v - unless - ( lastMay path == Just "recurseForDerivations" - && vt == Hercules.CNix.Expr.Raw.Bool - ) - do - logLocM DebugS $ - logStr $ - "Ignoring " - <> show path - <> " : " - <> (show vt :: Text) - -traverseSPWOs :: (StorePathWithOutputs -> IO ()) -> StdVector NixStorePathWithOutputs -> IO () -traverseSPWOs f v = do - v' <- Std.Vector.toListFP v - traverse_ f v' - --- | Documented in @docs/modules/ROOT/pages/legacy-evaluation.adoc@. -walk :: - (MonadUnliftIO m, KatipContext m) => - Store -> - Ptr EvalState -> - Value NixAttrs -> - RawValue -> - ConduitT i Event m () -walk store evalState = walk' True [] 10 - where - handleErrors path = Data.Conduit.handleC (yieldAttributeError path) - walk' :: - (MonadUnliftIO m, KatipContext m) => - -- If True, always walk this attribute set. Only True for the root. - Bool -> - -- Attribute path - [ByteString] -> - -- Depth of tree remaining - Integer -> - -- Auto arguments to pass to (attrset-)functions - Value NixAttrs -> - -- Current node of the walk - RawValue -> - -- Program that performs the walk and emits 'Event's - ConduitT i1 Event m () - walk' forceWalkAttrset path depthRemaining autoArgs v = - -- logLocM DebugS $ logStr $ "Walking " <> (show path :: Text) - handleErrors path $ - liftIO (match evalState v) - >>= \case - Left e -> - yieldAttributeError path e - Right m -> case m of - IsAttrs attrValue -> do - isDeriv <- liftIO $ isDerivation evalState v - if isDeriv - then walkDerivation store evalState True path attrValue - else do - walkAttrset <- - if forceWalkAttrset - then pure True - else -- Hydra doesn't seem to obey this, because it walks the - -- x64_64-linux etc attributes per package. Maybe those - -- are special cases? - -- For now, we will assume that people don't build a whole Nixpkgs - liftIO $ getRecurseForDerivations evalState attrValue - isfunctor <- liftIO $ isFunctor evalState v - if isfunctor && walkAttrset - then do - x <- liftIO (autoCallFunction evalState v autoArgs) - walk' True path (depthRemaining - 1) autoArgs x - else do - attrs <- liftIO $ getAttrs attrValue - void $ - flip M.traverseWithKey attrs $ - \name value -> - when (depthRemaining > 0 && walkAttrset) $ - walk' -- TODO: else warn - False - (path ++ [name]) - (depthRemaining - 1) - autoArgs - value - _any -> do - vt <- liftIO $ rawValueType v - unless - ( lastMay path - == Just "recurseForDerivations" - && vt - == Hercules.CNix.Expr.Raw.Bool - ) - $ logLocM DebugS $ - logStr $ - "Ignoring " - <> show path - <> " : " - <> (show vt :: Text) - pass - --- | Documented in @docs/modules/ROOT/pages/evaluation.adoc@. -walkDerivation :: - MonadIO m => - Store -> - Ptr EvalState -> - Bool -> - [ByteString] -> - Value NixAttrs -> - ConduitT i Event m () -walkDerivation store evalState effectsAnywhere path attrValue = do - drvStorePath <- getDrvFile evalState (rtValue attrValue) - drvPath <- liftIO $ CNix.storePathToPath store drvStorePath - typE <- runExceptT do - isEffect <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "isEffect") - case isEffect of - Just True - | effectsAnywhere - || inEffects path -> - throwE $ Right Attribute.Effect - Just True | otherwise -> throwE $ Left $ toException $ UserException "This derivation is marked as an effect, but effects are only allowed below the effects attribute." - _ -> pass - isDependenciesOnly <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "buildDependenciesOnly") - case isDependenciesOnly of - Just True -> throwE $ Right Attribute.DependenciesOnly - _ -> pass - phases <- liftEitherAs Left =<< liftIO (getAttrList evalState attrValue "phases") - case phases of - Just [aSingularPhase] -> - liftIO (match evalState aSingularPhase) >>= liftEitherAs Left >>= \case - IsString phaseNameValue -> do - phaseName <- liftIO (getStringIgnoreContext phaseNameValue) - when (phaseName == "nobuildPhase") do - throwE $ Right Attribute.DependenciesOnly - _ -> pass - _ -> pass - mayFail <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "ignoreFailure") - case mayFail of - Just True -> throwE $ Right Attribute.MayFail - _ -> pass - mustFail <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "requireFailure") - case mustFail of - Just True -> throwE $ Right Attribute.MustFail - _ -> pass - let yieldAttribute typ = - yield $ - Event.Attribute - Attribute.Attribute - { Attribute.path = path, - Attribute.drv = drvPath, - Attribute.typ = typ - } - case typE of - Left (Left e) -> yieldAttributeError path e - Left (Right t) -> yieldAttribute t - Right _ -> yieldAttribute Attribute.Regular - where - inEffects :: [ByteString] -> Bool - inEffects ("effects" : _) = True - inEffects _ = False - -liftEitherAs :: MonadError e m => (e0 -> e) -> Either e0 a -> m a -liftEitherAs f = liftEither . rmap - where - rmap (Left e) = Left (f e) - rmap (Right a) = Right a diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker/Build/Logger.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker/Build/Logger.hs --- a/hercules-ci-agent-worker/Hercules/Agent/Worker/Build/Logger.hs +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker/Build/Logger.hs @@ -265,6 +265,7 @@ filterProgress :: Monad m => ConduitT (Flush LogEntry) (Flush LogEntry) m () filterProgress = filterC \case Chunk LogEntry.Result {rtype = LogEntry.ResultTypeProgress} -> False + Chunk LogEntry.Result {rtype = LogEntry.ResultTypeSetExpected} -> False _ -> True nubProgress :: Monad m => ConduitT (Flush LogEntry) (Flush LogEntry) m () diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker/Conduit.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker/Conduit.hs new file mode 100644 --- /dev/null +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker/Conduit.hs @@ -0,0 +1,78 @@ +{-# LANGUAGE BlockArguments #-} + +module Hercules.Agent.Worker.Conduit where + +import Data.Conduit (ConduitT, await, awaitForever, yield, (.|)) +import Data.IORef (IORef, modifyIORef, newIORef, readIORef) +import qualified Data.Sequence as Seq +import Protolude hiding (pred, yield) + +tailC :: Monad m => Int -> ConduitT i i m () +tailC n = do + buf <- sinkTail n + for_ buf yield + +-- | Return the last @n@ items +sinkTail :: Monad m => Int -> ConduitT i o m (Seq i) +sinkTail n = do + doBuffer mempty + where + doBuffer st = + await >>= \case + Nothing -> pure st + Just item -> doBuffer $! (Seq.drop (length st - n + 1) st Seq.:|> item) + +-- | Take at most @n@ items that satisfy the predicate, then stop consuming, +-- even if the next item does not match the predicate. +-- +-- Return the number of counted messages and the total number of messages written. +takeCWhileStopEarly :: Monad m => (i -> Bool) -> Int -> ConduitT i i m (Int, Int) +takeCWhileStopEarly counts limit = go 0 0 + where + go counted total | counted >= limit = pure (counted, total) + go counted total = + await >>= \case + Nothing -> pure (counted, total) + Just item -> do + yield item + if item & counts + then go (counted + 1) (total + 1) + else go counted (total + 1) + +countProduction :: (Num n, MonadIO m) => (i -> Bool) -> IORef n -> ConduitT i i m () +countProduction pred counter = awaitForever (\i -> increment i *> yield i) + where + increment i | pred i = liftIO $ modifyIORef counter (+ 1) + increment _ = pass + +withInputProductionCount :: MonadIO m => (i -> Bool) -> ConduitT i o m a -> ConduitT i o m (Int, a) +withInputProductionCount pred conduit = do + counter <- liftIO $ newIORef 0 + r <- countProduction pred counter .| conduit + (,r) <$> liftIO (readIORef counter) + +withMessageLimit :: + MonadIO m => + (a -> Bool) -> + -- | First limit + Int -> + -- | Max tail part limit + Int -> + -- | What to do when truncatable output starts (waiting starts) + ConduitT a a m () -> + -- | What to do before yielding a truncated tail + (Int -> ConduitT a a m ()) -> + -- | What to do after yielding a truncated tail + (Int -> ConduitT a a m ()) -> + ConduitT a a m () +withMessageLimit pred firstLimit tailLimit afterFirst beforeTail afterTail = do + (c, _inclFlush) <- takeCWhileStopEarly pred firstLimit + when (c == firstLimit) afterFirst + (n, x) <- withInputProductionCount pred do + sinkTail tailLimit + let between = n - Seq.length x + when (between > 0) do + beforeTail between + for_ x yield + when (between > 0) do + afterTail between diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker/Env.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker/Env.hs new file mode 100644 --- /dev/null +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker/Env.hs @@ -0,0 +1,23 @@ +module Hercules.Agent.Worker.Env where + +import Control.Concurrent.STM (TVar) +import Data.IORef (IORef) +import Data.UUID (UUID) +import Hercules.Agent.Worker.HerculesStore (HerculesStore) +import qualified Hercules.Agent.WorkerProtocol.Command.BuildResult as BuildResult +import Hercules.Agent.WorkerProtocol.Event (Event) +import Hercules.CNix (Ref) +import Hercules.CNix.Store (Store, StorePath) +import Protolude + +data HerculesState = HerculesState + { drvsCompleted :: TVar (Map StorePath (UUID, BuildResult.BuildStatus)), + drvBuildAsyncs :: TVar (Map StorePath (Async (UUID, BuildResult.BuildStatus))), + drvRebuildAsyncs :: TVar (Map StorePath (Async (UUID, BuildResult.BuildStatus))), + drvOutputSubstituteAsyncs :: TVar (Map (StorePath, ByteString) (Async StorePath)), + drvsInProgress :: IORef (Set StorePath), + herculesStore :: Ptr (Ref HerculesStore), + wrappedStore :: Store, + shortcutChannel :: Chan (Maybe Event), + extraNixOptions :: [(Text, Text)] + } diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker/Error.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker/Error.hs new file mode 100644 --- /dev/null +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker/Error.hs @@ -0,0 +1,99 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE QuasiQuotes #-} +{-# LANGUAGE TemplateHaskell #-} + +module Hercules.Agent.Worker.Error where + +import Data.ByteString.Unsafe (unsafePackMallocCString) +import Foreign (alloca, peek) +import Hercules.CNix.Encapsulation (HasEncapsulation (moveToForeignPtrWrapper)) +import Hercules.CNix.Store (StorePath (StorePath), traverseNonNull) +import qualified Hercules.CNix.Store.Context +import qualified Language.C.Inline.Cpp as C +import qualified Language.C.Inline.Cpp.Exception as C +import Protolude + +C.context (C.cppCtx <> C.fptrCtx <> C.bsCtx <> Hercules.CNix.Store.Context.context) + +C.include "" +C.include "" +C.include "" +C.include "" +C.include "" + +type ANSIText = Text + +data ExceptionText = ExceptionText + { exceptionTextMessage :: ANSIText, + exceptionTextTrace :: Maybe ANSIText, + exceptionTextDerivationPath :: Maybe StorePath + } + +throwBuildError :: Text -> StorePath -> IO a +throwBuildError msg drv = do + let msgB = encodeUtf8 msg + [C.throwBlock| void { + std::string msg($bs-ptr:msgB, $bs-len:msgB); + nix::StorePath &drv = *$fptr-ptr:(nix::StorePath *drv); + throw hercules::HerculesBuildError(msg, drv); + }|] + panic "Could not throw HerculesBuildError!" -- tooling failure + +basicExceptionText :: ANSIText -> ExceptionText +basicExceptionText msg = + ExceptionText + { exceptionTextMessage = msg, + exceptionTextTrace = Nothing, + exceptionTextDerivationPath = Nothing + } + +renderException :: SomeException -> IO ExceptionText +renderException e | Just (C.CppStdException ex _msg _ty) <- fromException e = renderStdException ex +renderException e + | Just (C.CppNonStdException _ex maybeType) <- fromException e = + pure $ basicExceptionText $ "Unexpected C++ exception" <> foldMap (\t -> " of type " <> decodeUtf8With lenientDecode t) maybeType +renderException e | Just (FatalError msg) <- fromException e = pure $ basicExceptionText msg +renderException e = pure $ basicExceptionText $ toS $ displayException e + +renderStdException :: C.CppExceptionPtr -> IO ExceptionText +renderStdException e = alloca \traceStrPtr -> alloca \buildErrorDrvPtr -> do + msg <- + [C.throwBlock| const char * { + const char **traceStrPtr = $(const char **traceStrPtr); + *traceStrPtr = nullptr; + + nix::StorePath **buildErrorDrvPtr = $(nix::StorePath **buildErrorDrvPtr); + *buildErrorDrvPtr = nullptr; + + std::exception_ptr *e = $fptr-ptr:(std::exception_ptr *e); + try { + std::rethrow_exception(*e); + } catch (const hercules::HerculesBuildError &e) { + *buildErrorDrvPtr = new nix::StorePath(e.drv); + const char *msg; + hercules::copyErrorStrings(e, &msg, traceStrPtr); + return msg; + } catch (const nix::Error &e) { + const char *msg; + hercules::copyErrorStrings(e, &msg, traceStrPtr); + return msg; + } catch (const std::exception &e) { + return strdup(e.what()); + } catch (...) { + // shouldn't happen because inline-c-cpp only put std::exception in CppStdException + throw std::runtime_error("renderStdException: Attempt to render unknown exception."); + } + }|] + >>= unsafePackMallocCString + <&> decodeUtf8With lenientDecode + traceText <- + peek traceStrPtr >>= traverseNonNull \s -> + unsafePackMallocCString s <&> decodeUtf8With lenientDecode + drv <- + peek buildErrorDrvPtr >>= traverseNonNull moveToForeignPtrWrapper + pure $ + ExceptionText + { exceptionTextMessage = msg, + exceptionTextTrace = traceText, + exceptionTextDerivationPath = drv + } diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker/Evaluate.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker/Evaluate.hs new file mode 100644 --- /dev/null +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker/Evaluate.hs @@ -0,0 +1,699 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE OverloadedLabels #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-} + +{-# HLINT ignore "Use const" #-} + +module Hercules.Agent.Worker.Evaluate where + +import Conduit +import Control.Concurrent.Async (pollSTM) +import Control.Concurrent.STM hiding (check) +import Control.Exception.Safe (isAsyncException) +import Control.Monad.Except +import Control.Monad.IO.Unlift +import Data.Coerce (coerce) +import qualified Data.Conduit +import Data.Conduit.Katip.Orphans () +import Data.IORef +import qualified Data.Map as M +import qualified Data.Set as S +import Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent (OnPushHandlerEvent (OnPushHandlerEvent)) +import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent +import qualified Hercules.API.Agent.Evaluate.EvaluateTask as EvaluateTask +import qualified Hercules.API.Agent.Evaluate.EvaluateTask.OnPush as OnPush +import qualified Hercules.API.Agent.Evaluate.ImmutableGitInput as API.ImmutableGitInput +import qualified Hercules.API.Agent.Evaluate.ImmutableInput as API.ImmutableInput +import Hercules.Agent.NixFile (HerculesCISchema, getHerculesCI, homeExprRawValue, loadNixFile, parseExtraInputs) +import qualified Hercules.Agent.NixFile as NixFile +import Hercules.Agent.NixFile.HerculesCIArgs (CISystems (CISystems), HerculesCIMeta (HerculesCIMeta), fromGitSource) +import qualified Hercules.Agent.NixFile.HerculesCIArgs +import Hercules.Agent.Worker.Env (HerculesState (..)) +import Hercules.Agent.Worker.Error (ExceptionText (exceptionTextDerivationPath), exceptionTextMessage, exceptionTextTrace, renderException, throwBuildError) +import Hercules.Agent.Worker.HerculesStore (nixStore, setBuilderCallback) +import Hercules.Agent.Worker.STM (asyncInTVarMap) +import qualified Hercules.Agent.WorkerProtocol.Command.BuildResult as BuildResult +import Hercules.Agent.WorkerProtocol.Command.Eval + ( Eval, + ) +import qualified Hercules.Agent.WorkerProtocol.Command.Eval as Eval +import Hercules.Agent.WorkerProtocol.Event + ( Event, + ViaJSON (ViaJSON), + ) +import qualified Hercules.Agent.WorkerProtocol.Event as Event +import qualified Hercules.Agent.WorkerProtocol.Event.Attribute as Attribute +import qualified Hercules.Agent.WorkerProtocol.Event.AttributeError as AttributeError +import qualified Hercules.Agent.WorkerProtocol.Event.AttributeIFD as Event.AttributeIFD +import Hercules.CNix as CNix +import Hercules.CNix.Expr (Match (IsAttrs, IsString), NixAttrs, RawValue, addAllowedPath, addInternalAllowedPaths, autoCallFunction, evalArgs, getAttrBool, getAttrList, getAttrs, getDrvFile, getFlakeFromArchiveUrl, getFlakeFromGit, getRecurseForDerivations, getStringIgnoreContext, isDerivation, isFunctor, match, rawValueType, rtValue, toRawValue, toValue, withEvalStateConduit) +import Hercules.CNix.Expr.Context (EvalState) +import qualified Hercules.CNix.Expr.Raw +import Hercules.CNix.Expr.Schema (MonadEval, PSObject, dictionaryToMap, fromPSObject, requireDict, (#.), (#?), (#?!), ($?)) +import qualified Hercules.CNix.Expr.Schema as Schema +import Hercules.CNix.Expr.Typed (Value) +import Hercules.CNix.Std.Vector (StdVector) +import qualified Hercules.CNix.Std.Vector as Std.Vector +import Hercules.CNix.Store.Context (NixStorePathWithOutputs) +import Hercules.Error +import Hercules.UserException (UserException (UserException)) +import Katip +import Protolude hiding (bracket, catch, check, evalState, wait, withAsync, yield) +import qualified UnliftIO +import UnliftIO.Async (wait) +import UnliftIO.Exception (bracket, catch) +import Prelude () + +data AsyncRealisationRequired = AsyncRealisationRequired + { realisationRequiredDerivationPath :: StorePath, + realisationRequiredOutputName :: ByteString + } + deriving (Show, Typeable, Eq, Ord) + +instance Exception AsyncRealisationRequired + +isAsyncRealisationRequired :: SomeException -> Bool +isAsyncRealisationRequired e = case fromException e of + Just (_ :: AsyncRealisationRequired) -> True + Nothing -> False + +data EvalEnv = EvalEnv + { evalEnvHerculesState :: HerculesState, + evalEnvState :: Ptr EvalState, + evalEnvIsNonBlocking :: IORef Bool + } + +evalEnvStore :: EvalEnv -> Store +evalEnvStore = nixStore . herculesStore . evalEnvHerculesState + +-- TODO: test +autoArgArgs :: Map Text Eval.Arg -> [ByteString] +autoArgArgs kvs = do + (k, v) <- M.toList kvs + case v of + Eval.LiteralArg s -> ["--argstr", encodeUtf8 k, s] + Eval.ExprArg s -> ["--arg", encodeUtf8 k, s] + +-- Ensure that a nested build invocation does not happen by a mistake in wiring. +-- +-- (It does not happen, but this de-escalates a potential recursion bug to just +-- an error.) +withDrvInProgress :: MonadUnliftIO m => HerculesState -> StorePath -> m a -> m a +withDrvInProgress HerculesState {drvsInProgress = ref} drvPath = + bracket acquire release . const + where + acquire = + liftIO $ + join $ + atomicModifyIORef ref $ \inprg -> + if drvPath `S.member` inprg + then (inprg, throwIO $ FatalError "Refusing to build derivation that should have been built remotely. Presumably, substitution has failed.") + else (S.insert drvPath inprg, pass) + release _ = + liftIO $ + atomicModifyIORef ref $ \inprg -> + (S.delete drvPath inprg, ()) + +anyAlternative :: (Foldable l, Alternative f) => l a -> f a +anyAlternative = getAlt . foldMap (Alt . pure) + +yieldAttributeError :: MonadIO m => Store -> [ByteString] -> SomeException -> ConduitT i Event m () +yieldAttributeError store path e = do + exceptionText <- liftIO $ renderException e + drvPath <- liftIO $ traverse (storePathToPath store) (exceptionTextDerivationPath exceptionText) + yield $ + Event.AttributeError $ + AttributeError.AttributeError + { AttributeError.path = path, + AttributeError.message = exceptionTextMessage exceptionText, + AttributeError.errorDerivation = drvPath <&> decodeUtf8With lenientDecode, + AttributeError.errorType = if isJust (exceptionTextDerivationPath exceptionText) then Just "BuildException" else Just (show (typeOf e)), + AttributeError.trace = exceptionTextTrace exceptionText + } + +maybeThrowBuildException :: MonadIO m => BuildResult.BuildStatus -> StorePath -> m () +maybeThrowBuildException result drv = + case result of + BuildResult.Failure -> liftIO $ throwBuildError ("Could not build derivation " <> show drv <> ", which is required during evaluation.") drv + BuildResult.DependencyFailure -> liftIO $ throwBuildError ("Could not build a dependency of derivation " <> show drv <> ", which is required during evaluation.") drv + BuildResult.Success -> pass + +mkCache :: forall k a m. (MonadUnliftIO m, Ord k) => (SomeException -> Bool) -> IO (k -> m a -> m a) +mkCache cacheableException = do + doneRef <- liftIO $ newIORef mempty + let hasBeenBuilt :: k -> m (Maybe (Either SomeException a)) + hasBeenBuilt p = liftIO $ readIORef doneRef <&> \c -> M.lookup p c + cacheBy p io = do + b <- hasBeenBuilt p + case b of + Just x -> liftIO $ escalate x + Nothing -> do + r <- UnliftIO.tryAny io + liftIO do + let cacheable = case r of + Right {} -> True + Left e -> cacheableException e + when cacheable do + modifyIORef doneRef (M.insert p (r :: Either SomeException a)) + escalate r + pure cacheBy + +runEval :: + forall i m. + (MonadResource m, KatipContext m, MonadUnliftIO m, MonadThrow m) => + HerculesState -> + Eval -> + ConduitM i Event m () +runEval st@HerculesState {herculesStore = hStore, shortcutChannel = shortcutChan, drvsCompleted = drvsCompl} eval = do + for_ (Eval.extraNixOptions eval) $ liftIO . uncurry setGlobalOption + for_ (Eval.extraNixOptions eval) $ liftIO . uncurry setOption + let store = nixStore hStore + isFlake = Eval.isFlakeJob eval + s <- storeUri store + UnliftIO unlift <- lift askUnliftIO + let decode = decodeUtf8With lenientDecode + + cachingBuilt <- liftIO $ mkCache (not . isAsyncRealisationRequired) + + isNonBlocking <- liftIO (newIORef False) + + liftIO . setBuilderCallback hStore $ + traverseSPWOs $ \storePathWithOutputs -> unlift $ do + drvStorePath <- liftIO $ getStorePath storePathWithOutputs + drvPath <- liftIO $ CNix.storePathToPath store drvStorePath + cachingBuilt drvPath do + let pathText = decode drvPath + outputs <- liftIO $ getOutputs storePathWithOutputs + katipAddContext (sl "fullpath" pathText) $ + for_ outputs $ \outputName -> do + withDrvInProgress st drvStorePath $ do + derivation <- liftIO $ getDerivation store drvStorePath + drvName <- liftIO $ getDerivationNameFromPath drvStorePath + drvOutputs <- liftIO $ getDerivationOutputs store drvName derivation + outputPath <- + case find (\o -> derivationOutputName o == outputName) drvOutputs of + Nothing -> panic $ "output " <> show outputName <> " does not exist on " <> pathText + Just o -> case derivationOutputPath o of + Just x -> pure x + Nothing -> + -- FIXME ca-derivations + panic $ "output path unknown for output " <> show outputName <> " on " <> pathText <> ". ca-derivations is not supported yet." + liftIO $ addTemporaryRoot store outputPath + + isValid <- liftIO $ isValidPath store outputPath + if isValid + then do + logLocM DebugS "Output already valid" + -- Report IFD + liftIO $ writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) Nothing False + else do + don'tBlock <- liftIO (readIORef isNonBlocking) + let doBlock = not don'tBlock + + liftIO $ writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) Nothing doBlock + + buildAsync <- liftIO $ asyncInTVarMap (drvStorePath, outputName) (drvOutputSubstituteAsyncs st) do + ensurePath (wrappedStore st) outputPath + `catch` \(_e :: SomeException) -> do + -- wait for build that was requested before + + (attempt0, result) <- wait <=< liftIO $ asyncInTVarMap drvStorePath (drvBuildAsyncs st) do + atomically $ do + c <- readTVar drvsCompl + anyAlternative $ M.lookup drvStorePath c + + maybeThrowBuildException result drvStorePath + clearSubstituterCaches + clearPathInfoCache store + ensurePath (wrappedStore st) outputPath `catch` \(_e1 :: SomeException) -> do + writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) (Just attempt0) doBlock + + (_, result') <- + wait =<< asyncInTVarMap drvStorePath (drvRebuildAsyncs st) do + atomically $ do + c <- readTVar drvsCompl + (attempt1, r) <- anyAlternative $ M.lookup drvStorePath c + guard (attempt1 /= attempt0) + pure (attempt1, r) + + maybeThrowBuildException result' drvStorePath + clearSubstituterCaches + clearPathInfoCache store + ensurePath (wrappedStore st) outputPath `catch` \e2 -> + liftIO $ + throwBuildError + ( "Build failure: output could not be retrieved on the evaluating agent, despite a successful rebuild. Exception: " + <> show (e2 :: SomeException) + ) + drvStorePath + pure outputPath + + maybeAlreadyDone <- liftIO $ poll buildAsync + case maybeAlreadyDone of + Just r -> void $ escalate r + Nothing -> do + if don'tBlock + then throwIO (AsyncRealisationRequired drvStorePath outputName) + else void $ wait buildAsync + + withEvalStateConduit store $ \evalState -> do + let evalEnv :: EvalEnv + evalEnv = + EvalEnv + { evalEnvHerculesState = st, + evalEnvIsNonBlocking = isNonBlocking, + evalEnvState = evalState + } + + liftIO do + addInternalAllowedPaths evalState + for_ (Eval.allowedPaths eval) (addAllowedPath evalState) + katipAddContext (sl "storeURI" (decode s)) $ + logLocM DebugS "EvalState loaded." + args <- + liftIO $ + evalArgs evalState (autoArgArgs (Eval.autoArguments eval)) + Data.Conduit.handleC (yieldAttributeError store []) $ + do + homeExpr <- getHomeExpr evalState eval + let hargs = fromGitSource (coerce $ Eval.gitSource eval) meta + meta = HerculesCIMeta {apiBaseUrl = Eval.apiBaseUrl eval, ciSystems = CISystems (Eval.ciSystems eval)} + liftIO (flip runReaderT evalState $ getHerculesCI homeExpr hargs) >>= \case + Nothing -> + -- legacy + walk evalEnv args (homeExprRawValue homeExpr) + Just herculesCI -> do + case Event.fromViaJSON (Eval.selector eval) of + EvaluateTask.ConfigOrLegacy -> do + yield Event.JobConfig + sendConfig evalState isFlake herculesCI + EvaluateTask.OnPush onPush -> + transPipe (`runReaderT` evalState) do + walkOnPush evalEnv onPush herculesCI + yield Event.EvaluationDone + +getHomeExpr :: (MonadThrow m, MonadIO m) => Ptr EvalState -> Eval -> m NixFile.HomeExpr +getHomeExpr evalState eval = + if Eval.isFlakeJob eval + then + NixFile.Flake <$> liftIO do + srcInput <- case Eval.srcInput eval of + Just x -> pure x + Nothing -> panic "srcInput is required for flake job" + raw <- mkImmutableGitInputFlakeThunk evalState (Event.fromViaJSON srcInput) + let pso :: PSObject (Schema.Attrs '[]) + pso = Schema.PSObject {value = raw, provenance = Schema.Other "flake.nix"} + toValue evalState pso + else escalateAs UserException =<< liftIO (loadNixFile evalState (toS $ Eval.cwd eval) (coerce $ Eval.gitSource eval)) + +walkOnPush :: (MonadEval m, MonadUnliftIO m, KatipContext m, MonadThrow m) => EvalEnv -> OnPush.OnPush -> PSObject HerculesCISchema -> ConduitT i Event m () +walkOnPush evalEnv onPushParams herculesCI = do + let evalState = evalEnvState evalEnv + onPushHandler <- herculesCI #?! #onPush >>= requireDict (OnPush.name onPushParams) + inputs <- liftIO $ do + inputs <- for (OnPush.inputs onPushParams) \input -> do + inputToValue evalState input + toRawValue evalState inputs + outputsFun <- onPushHandler #. #outputs + outputs <- outputsFun $? (Schema.PSObject {value = inputs, provenance = Schema.Data}) + simpleWalk evalEnv (Schema.value outputs) + +inputToValue :: Ptr EvalState -> API.ImmutableInput.ImmutableInput -> IO RawValue +inputToValue evalState (API.ImmutableInput.ArchiveUrl u) = getFlakeFromArchiveUrl evalState u +inputToValue evalState (API.ImmutableInput.Git g) = mkImmutableGitInputFlakeThunk evalState g + +mkImmutableGitInputFlakeThunk :: Ptr EvalState -> API.ImmutableGitInput.ImmutableGitInput -> IO RawValue +mkImmutableGitInputFlakeThunk evalState git = do + -- TODO: allow picking ssh/http url + getFlakeFromGit + evalState + (API.ImmutableGitInput.httpURL git) + (API.ImmutableGitInput.ref git) + (API.ImmutableGitInput.rev git) + +sendConfig :: MonadIO m => Ptr EvalState -> Bool -> PSObject HerculesCISchema -> ConduitT i Event m () +sendConfig evalState isFlake herculesCI = flip runReaderT evalState $ do + herculesCI #? #onPush >>= traverse_ \onPushes -> do + attrs <- dictionaryToMap onPushes + for_ (M.mapWithKey (,) attrs) \(name, onPush) -> do + ei <- onPush #? #extraInputs >>= traverse parseExtraInputs + enable <- onPush #? #enable >>= traverse fromPSObject <&> fromMaybe True + when enable . lift . yield . Event.OnPushHandler . ViaJSON $ + OnPushHandlerEvent + { handlerName = decodeUtf8 name, + handlerExtraInputs = M.mapKeys decodeUtf8 (fromMaybe mempty ei), + isFlake = isFlake + } + +data TreeWork = TreeWork + { treeWorkAttrPath :: [ByteString], + -- Redundant but convenient + treeWorkThunk :: RawValue, + -- Redundant but convenient + treeWorkRemainingDepth :: Int + } + +type Enqueue i m o = i -> (o -> m ()) -> IO () + +pollingWith :: + forall i m o. + (Ord i, MonadIO m) => + -- | A 'finally'-like function that powers an assertion. + -- + -- Not used for resource allocation; does not have to be perfect. + (m () -> IO () -> m ()) -> + -- | Arg will not be empty. Return must not be empty. + (forall x. Map i x -> m (Map i o)) -> + -- | Do work. Enqueue function can be called from anywhere, but will be ignored after the returned action is done. + ((i -> (o -> m ()) -> IO ()) -> m ()) -> + -- | Iterate until work done. + m () +pollingWith ffinally poller startWork = do + q <- liftIO (newIORef mempty) + operational <- liftIO (newIORef True) + + let checkOperational = do + op <- readIORef operational + when (not op) (panic "pollingWith has stopped. Can not continue.") + + enqueue :: Enqueue i m o + enqueue i f = do + checkOperational + modifyIORef' q $ + M.alter + \case + Nothing -> Just f + Just f0 -> Just (\o -> f0 o *> f o) + i + + loop :: m () + loop = do + -- peek all, restore blocked next + q0 <- liftIO (atomicModifyIORef' q (mempty,)) + if null q0 + then pass + else do + justCompleted <- poller q0 + let blocked = M.difference q0 justCompleted + + when (null justCompleted) do + panic "pollingWith: poller must not return empty" + + liftIO (atomicModifyIORef' q \q1 -> (M.unionWith (liftA2 (>>)) blocked q1, ())) + + sequenceA_ $ M.intersectionWith ($) q0 justCompleted + + loop + + startWork enqueue + loop `ffinally` writeIORef operational False + +handleExceptions :: + (Exception t, MonadUnliftIO m) => + Store -> + ( [ByteString] -> + t -> + (Either SomeException b -> ConduitT i Event m ()) -> + ConduitT i Event m () + ) -> + [ByteString] -> + ConduitT i Event m () -> + ConduitT i Event m () +handleExceptions store enqueue path m = go + where + go = + -- not catchC or handleC, to make sure async exceptions aren't masked + tryC m >>= \case + Left e -> handler e + Right r -> pure r + + handler e | isAsyncException e = throwIO e + handler e = case fromException e of + Just rr -> do + enqueue path rr \case + Right {} -> go + Left exception -> + -- TODO: throw a Nix native exception in the builder callback instead? + -- trace would have to be preserved from the first execution + yieldAttributeError store path exception + Nothing -> do + yieldAttributeError store path e + +-- | Documented in @docs/modules/ROOT/pages/evaluation.adoc@. +simpleWalk :: + (MonadUnliftIO m, KatipContext m) => + EvalEnv -> + RawValue -> + ConduitT i Event m () +simpleWalk evalEnv initialThunk = do + let store = evalEnvStore evalEnv + evalState = evalEnvState evalEnv + + walk' :: + (MonadUnliftIO m, KatipContext m) => + ([ByteString] -> AsyncRealisationRequired -> (Either SomeException () -> ConduitT i1 Event m ()) -> ConduitT i1 Event m ()) -> + TreeWork -> + -- Program that performs the walk and emits 'Event's + ConduitT i1 Event m () + walk' enqueue treeWork = do + let path = treeWorkAttrPath treeWork + depthRemaining = treeWorkRemainingDepth treeWork + v = treeWorkThunk treeWork + + handleExceptions store enqueue path do + m <- liftIO $ escalate =<< match evalState v + case m of + IsAttrs attrValue -> do + isDeriv <- liftIO $ isDerivation evalState v + if isDeriv + then walkDerivation store evalState False path attrValue + else do + attrs <- liftIO $ getAttrs attrValue + void $ + flip M.traverseWithKey attrs $ + \name value -> + if depthRemaining > 0 + then + walk' + enqueue + TreeWork + { treeWorkRemainingDepth = depthRemaining - 1, + treeWorkAttrPath = path ++ [name], + treeWorkThunk = value + } + else yield (Event.Error $ "Max recursion depth reached at path " <> show path) + _any -> do + vt <- liftIO $ rawValueType v + unless + ( lastMay path == Just "recurseForDerivations" + && vt == Hercules.CNix.Expr.Raw.Bool + ) + do + logLocM DebugS $ logStr $ "Ignoring " <> show path <> " : " <> (show vt :: Text) + + withIFDQueue evalEnv \enqueue -> + walk' + enqueue + TreeWork + { treeWorkAttrPath = [], + treeWorkRemainingDepth = 10, + treeWorkThunk = initialThunk + } + +withIFDQueue :: + MonadUnliftIO m => + EvalEnv -> + ( ([ByteString] -> AsyncRealisationRequired -> (Either SomeException () -> ConduitT i Event m ()) -> ConduitT i Event m ()) -> + ConduitT i Event m () + ) -> + ConduitT i Event m () +withIFDQueue evalEnv doIt = do + let poller :: + MonadIO m => + Map (StorePath, ByteString) x -> + ConduitT i Event m (Map (StorePath, ByteString) (Either SomeException ())) + poller q = do + liftIO $ atomically do + allBuilds <- readTVar (drvOutputSubstituteAsyncs (evalEnvHerculesState evalEnv)) + x <- + q & M.traverseWithKey \qKey _ -> do + case M.lookup qKey allBuilds of + Nothing -> panic "A realisation should have been started before throwing AsyncRealisationRequired" + Just asy -> pollSTM asy + let completed = void <$> M.mapMaybe identity x + guard (not (null completed)) + pure completed + + bestEffortFinally f ff = catchC f (\e -> liftIO ff >> throwIO (e :: SomeException)) + + betterEnqueue enqueue attrPath rr m = do + drvPath <- liftIO $ storePathToPath (evalEnvStore evalEnv) $ realisationRequiredDerivationPath rr + let ev = + Event.AttributeIFD.AttributeIFD + { path = attrPath, + derivationPath = drvPath, + derivationOutput = realisationRequiredOutputName rr, + done = False + } + yield (Event.AttributeIFD ev) + liftIO $ enqueue (realisationRequiredDerivationPath rr, realisationRequiredOutputName rr) \outcome -> do + yield (Event.AttributeIFD $ ev {Event.AttributeIFD.done = True}) + m outcome + + reset <- liftIO $ readIORef (evalEnvIsNonBlocking evalEnv) + liftIO $ writeIORef (evalEnvIsNonBlocking evalEnv) True + + pollingWith bestEffortFinally poller (doIt . betterEnqueue) + + liftIO $ writeIORef (evalEnvIsNonBlocking evalEnv) reset + +traverseSPWOs :: (StorePathWithOutputs -> IO ()) -> StdVector NixStorePathWithOutputs -> IO () +traverseSPWOs f v = do + v' <- Std.Vector.toListFP v + traverse_ f v' + +-- | Documented in @docs/modules/ROOT/pages/legacy-evaluation.adoc@. +walk :: + forall i m. + (MonadUnliftIO m, KatipContext m) => + EvalEnv -> + Value NixAttrs -> + RawValue -> + ConduitT i Event m () +walk evalEnv autoArgs v0 = withIFDQueue evalEnv \enqueue -> + let start = walk' True [] 10 v0 + store = evalEnvStore evalEnv + evalState = evalEnvState evalEnv + walk' :: + (MonadUnliftIO m, KatipContext m) => + -- If True, always walk this attribute set. Only True for the root. + Bool -> + -- Attribute path + [ByteString] -> + -- Depth of tree remaining + Integer -> + -- Current node of the walk + RawValue -> + -- Program that performs the walk and emits 'Event's + ConduitT i Event m () + walk' forceWalkAttrset path depthRemaining v = + -- logLocM DebugS $ logStr $ "Walking " <> (show path :: Text) + handleExceptions store enqueue path $ + liftIO (match evalState v) + >>= \case + Left e -> + throwIO e + Right m -> case m of + IsAttrs attrValue -> do + isDeriv <- liftIO $ isDerivation evalState v + if isDeriv + then walkDerivation store evalState True path attrValue + else do + walkAttrset <- + if forceWalkAttrset + then pure True + else -- Hydra doesn't seem to obey this, because it walks the + -- x64_64-linux etc attributes per package. Maybe those + -- are special cases? + -- For now, we will assume that people don't build a whole Nixpkgs + liftIO $ getRecurseForDerivations evalState attrValue + isfunctor <- liftIO $ isFunctor evalState v + if isfunctor && walkAttrset + then do + x <- liftIO (autoCallFunction evalState v autoArgs) + walk' True path (depthRemaining - 1) x + else do + attrs <- liftIO $ getAttrs attrValue + void $ + flip M.traverseWithKey attrs $ + \name value -> + when (depthRemaining > 0 && walkAttrset) $ + walk' -- TODO: else warn + False + (path ++ [name]) + (depthRemaining - 1) + value + _any -> do + vt <- liftIO $ rawValueType v + unless + ( lastMay path + == Just "recurseForDerivations" + && vt + == Hercules.CNix.Expr.Raw.Bool + ) + $ logLocM DebugS $ + logStr $ + "Ignoring " + <> show path + <> " : " + <> (show vt :: Text) + pass + in start + +-- | Documented in @docs/modules/ROOT/pages/evaluation.adoc@. +walkDerivation :: + MonadIO m => + Store -> + Ptr EvalState -> + Bool -> + [ByteString] -> + Value NixAttrs -> + ConduitT i Event m () +walkDerivation store evalState effectsAnywhere path attrValue = do + drvStorePath <- getDrvFile evalState (rtValue attrValue) + drvPath <- liftIO $ CNix.storePathToPath store drvStorePath + typE <- runExceptT do + isEffect <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "isEffect") + case isEffect of + Just True + | effectsAnywhere + || inEffects path -> + throwE $ Right Attribute.Effect + Just True | otherwise -> throwE $ Left $ toException $ UserException "This derivation is marked as an effect, but effects are only allowed below the effects attribute." + _ -> pass + isDependenciesOnly <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "buildDependenciesOnly") + case isDependenciesOnly of + Just True -> throwE $ Right Attribute.DependenciesOnly + _ -> pass + phases <- liftEitherAs Left =<< liftIO (getAttrList evalState attrValue "phases") + case phases of + Just [aSingularPhase] -> + liftIO (match evalState aSingularPhase) >>= liftEitherAs Left >>= \case + IsString phaseNameValue -> do + phaseName <- liftIO (getStringIgnoreContext phaseNameValue) + when (phaseName == "nobuildPhase") do + throwE $ Right Attribute.DependenciesOnly + _ -> pass + _ -> pass + mayFail <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "ignoreFailure") + case mayFail of + Just True -> throwE $ Right Attribute.MayFail + _ -> pass + mustFail <- liftEitherAs Left =<< liftIO (getAttrBool evalState attrValue "requireFailure") + case mustFail of + Just True -> throwE $ Right Attribute.MustFail + _ -> pass + let yieldAttribute typ = + yield $ + Event.Attribute + Attribute.Attribute + { Attribute.path = path, + Attribute.drv = drvPath, + Attribute.typ = typ + } + case typE of + Left (Left e) -> yieldAttributeError store path e + Left (Right t) -> yieldAttribute t + Right _ -> yieldAttribute Attribute.Regular + where + inEffects :: [ByteString] -> Bool + inEffects ("effects" : _) = True + inEffects _ = False + +liftEitherAs :: MonadError e m => (e0 -> e) -> Either e0 a -> m a +liftEitherAs f = liftEither . rmap + where + rmap (Left e) = Left (f e) + rmap (Right a) = Right a diff --git a/hercules-ci-agent-worker/Hercules/Agent/Worker/STM.hs b/hercules-ci-agent-worker/Hercules/Agent/Worker/STM.hs new file mode 100644 --- /dev/null +++ b/hercules-ci-agent-worker/Hercules/Agent/Worker/STM.hs @@ -0,0 +1,48 @@ +{-# LANGUAGE BlockArguments #-} + +module Hercules.Agent.Worker.STM where + +import Control.Concurrent.Async +import Control.Concurrent.STM (TVar, readTVar, writeTVar) +import Control.Monad.Fix (mfix) +import qualified Data.Map as M +import Protolude + +asyncIfSTM :: + -- | Either return an existing async or return @b@ as an input to a new one. + -- + -- The returned transaction must be not evaluate the 'Async' argument, + -- but can write it somewhere. + (Async a -> STM (Either (Async a) b)) -> + -- | The action to run when the STM transaction decided to create a new 'Async'. + (b -> IO a) -> + -- | Either the async returned by the STM transaction, or a new async waiting + -- for the provided IO. + IO (Async a) +asyncIfSTM cond io = mfix \lazyAsync -> + atomically (cond lazyAsync) >>= \case + Left early -> pure early + Right b -> do + async (io b) + +ensureTVarMapItem :: Ord k => k -> STM v -> TVar (Map k v) -> STM (Bool, v) +ensureTVarMapItem key mkValue mapVar = do + map0 <- readTVar mapVar + case M.lookup key map0 of + Nothing -> do + value <- mkValue + writeTVar mapVar (M.insert key value map0) + pure (False, value) + Just value0 -> do + pure (True, value0) + +asyncInTVarMap :: Ord k => k -> TVar (Map k (Async a)) -> IO a -> IO (Async a) +asyncInTVarMap key mapVar action = + asyncIfSTM + ( \asy -> do + (existed, v) <- ensureTVarMapItem key (pure asy) mapVar + if existed + then pure (Left v) + else pure (Right ()) + ) + (\() -> action) diff --git a/hercules-ci-agent.cabal b/hercules-ci-agent.cabal --- a/hercules-ci-agent.cabal +++ b/hercules-ci-agent.cabal @@ -1,7 +1,7 @@ cabal-version: 2.4 name: hercules-ci-agent -version: 0.9.3 +version: 0.9.4 synopsis: Runs Continuous Integration tasks on your machines category: Nix, CI, Testing, DevOps homepage: https://docs.hercules-ci.com @@ -93,6 +93,7 @@ Hercules.Agent.WorkerProtocol.Event Hercules.Agent.WorkerProtocol.Event.Attribute Hercules.Agent.WorkerProtocol.Event.AttributeError + Hercules.Agent.WorkerProtocol.Event.AttributeIFD Hercules.Agent.WorkerProtocol.Event.BuildResult Hercules.Agent.WorkerProtocol.LogSettings Hercules.Agent.WorkerProtocol.Orphans @@ -216,8 +217,9 @@ , exceptions , filepath , hercules-ci-agent + , hercules-ci-api , hercules-ci-api-core == 0.1.4.0 - , hercules-ci-api-agent == 0.4.4.0 + , hercules-ci-api-agent == 0.4.5.0 , hostname , http-client , http-client-tls @@ -267,14 +269,20 @@ Hercules.Agent.Worker.Build.Prefetched Hercules.Agent.Worker.Build.Logger Hercules.Agent.Worker.Build.Logger.Context + Hercules.Agent.Worker.Conduit Hercules.Agent.Worker.Effect + Hercules.Agent.Worker.Env + Hercules.Agent.Worker.Error + Hercules.Agent.Worker.Evaluate Hercules.Agent.Worker.HerculesStore Hercules.Agent.Worker.HerculesStore.Context Hercules.Agent.Worker.Logging + Hercules.Agent.Worker.STM hs-source-dirs: hercules-ci-agent-worker cxx-sources: + cbits/hercules-error.cxx cbits/hercules-logger.cxx cbits/nix-2.4/hercules-store.cxx @@ -356,6 +364,10 @@ Hercules.Agent.Nix.RetrieveDerivationInfoSpec Hercules.Agent.WorkerProcess Hercules.Agent.WorkerProcessSpec + Hercules.Agent.Worker.Conduit + Hercules.Agent.Worker.ConduitSpec + Hercules.Agent.Worker.STM + Hercules.Agent.Worker.STMSpec Hercules.Secrets Hercules.SecretsSpec Spec @@ -363,6 +375,7 @@ src test hercules-ci-agent + hercules-ci-agent-worker default-extensions: DeriveGeneric DeriveTraversable DisambiguateRecordFields FlexibleContexts InstanceSigs LambdaCase MultiParamTypeClasses NoImplicitPrelude OverloadedStrings RankNTypes TupleSections TypeApplications TypeOperators ghc-options: -Werror=incomplete-patterns -Werror=missing-fields -Wall -fwarn-tabs -fwarn-unused-imports -fwarn-missing-signatures -fwarn-name-shadowing -fwarn-incomplete-patterns -threaded -rtsopts -with-rtsopts=-N build-depends: @@ -390,6 +403,7 @@ , process , protolude , safe-exceptions + , stm , tagged , temporary , text diff --git a/hercules-ci-agent/Hercules/Agent/Build.hs b/hercules-ci-agent/Hercules/Agent/Build.hs --- a/hercules-ci-agent/Hercules/Agent/Build.hs +++ b/hercules-ci-agent/Hercules/Agent/Build.hs @@ -1,5 +1,6 @@ module Hercules.Agent.Build where +import qualified Data.Aeson as A import Data.IORef.Lifted import qualified Data.Map as M import qualified Hercules.API.Agent.Build as API.Build @@ -84,7 +85,15 @@ }, materializeDerivation = materialize } - exitCode <- runWorker procSpec (stderrLineHandler "Builder") commandChan writeEvent + let stderrHandler = + stderrLineHandler + ( M.fromList + [ ("taskId", A.toJSON (BuildTask.id buildTask)), + ("derivationPath", A.toJSON (BuildTask.derivationPath buildTask)) + ] + ) + "Builder" + exitCode <- runWorker procSpec stderrHandler commandChan writeEvent logLocM DebugS $ "Worker exit: " <> logStr (show exitCode :: Text) case exitCode of ExitSuccess -> pass diff --git a/hercules-ci-agent/Hercules/Agent/Cachix/Init.hs b/hercules-ci-agent/Hercules/Agent/Cachix/Init.hs --- a/hercules-ci-agent/Hercules/Agent/Cachix/Init.hs +++ b/hercules-ci-agent/Hercules/Agent/Cachix/Init.hs @@ -56,7 +56,7 @@ where toNetrcLine (name, keys) = do pt <- toList $ CachixCache.authToken keys - pure $ "machine " <> name <> ".cachix.org" <> " password " <> pt + pure $ "machine " <> name <> ".cachix.org" <> " login authtoken password " <> pt toPushCaches :: Map Text CachixCache.CachixCache -> IO (Map Text PushCache) toPushCaches = sequenceA . M.mapMaybeWithKey toPushCaches' diff --git a/hercules-ci-agent/Hercules/Agent/Effect.hs b/hercules-ci-agent/Hercules/Agent/Effect.hs --- a/hercules-ci-agent/Hercules/Agent/Effect.hs +++ b/hercules-ci-agent/Hercules/Agent/Effect.hs @@ -1,6 +1,8 @@ module Hercules.Agent.Effect where +import qualified Data.Aeson as A import Data.IORef +import qualified Data.Map as M import qualified Hercules.API.Agent.Effect.EffectTask as EffectTask import Hercules.API.TaskStatus (TaskStatus) import qualified Hercules.API.TaskStatus as TaskStatus @@ -82,7 +84,15 @@ isDefaultBranch = EffectTask.isDefaultBranch effectTask } } - exitCode <- runWorker procSpec (stderrLineHandler "Effect worker") commandChan writeEvent + let stderrHandler = + stderrLineHandler + ( M.fromList + [ ("taskId", A.toJSON (EffectTask.id effectTask)), + ("derivationPath", A.toJSON (EffectTask.derivationPath effectTask)) + ] + ) + "Effect worker" + exitCode <- runWorker procSpec stderrHandler commandChan writeEvent logLocM DebugS $ "Worker exit: " <> logStr (show exitCode :: Text) let showSig n | n == PS.sigABRT = " (Aborted)" showSig n | n == PS.sigBUS = " (Bus)" diff --git a/hercules-ci-agent/Hercules/Agent/Evaluate.hs b/hercules-ci-agent/Hercules/Agent/Evaluate.hs --- a/hercules-ci-agent/Hercules/Agent/Evaluate.hs +++ b/hercules-ci-agent/Hercules/Agent/Evaluate.hs @@ -12,6 +12,7 @@ import Control.Concurrent.Chan.Lifted import Control.Exception.Lifted (finally) import Control.Lens (at, (^?)) +import Control.Monad.IO.Unlift (askUnliftIO, unliftIO) import qualified Data.Aeson as A import Data.Aeson.Lens (_String) import qualified Data.ByteString.Lazy as BL @@ -26,6 +27,7 @@ import qualified Data.Set as S import qualified Data.Text as T import Data.UUID (UUID) +import Hercules.API (Id) import Hercules.API.Agent.Evaluate ( getDerivationStatus2, tasksUpdateEvaluation, @@ -34,6 +36,7 @@ import qualified Hercules.API.Agent.Evaluate.EvaluateEvent as EvaluateEvent import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.AttributeErrorEvent as AttributeErrorEvent import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.AttributeEvent as AttributeEvent +import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.AttributeIFDEvent as AttributeIFDEvent import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.BuildRequest as BuildRequest import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.BuildRequired as BuildRequired import qualified Hercules.API.Agent.Evaluate.EvaluateEvent.DerivationInfo as DerivationInfo @@ -45,6 +48,7 @@ import qualified Hercules.API.Agent.Evaluate.ImmutableGitInput as ImmutableGitInput import qualified Hercules.API.Agent.Evaluate.ImmutableInput as ImmutableInput import Hercules.API.Servant (noContent) +import Hercules.API.Task (Task) import qualified Hercules.Agent.Cache as Agent.Cache import qualified Hercules.Agent.Cachix.Env as Cachix.Env import qualified Hercules.Agent.Client @@ -62,6 +66,7 @@ ) import Hercules.Agent.NixFile (findNixFile) import Hercules.Agent.NixFile.GitSource (fromRefRevPath) +import qualified Hercules.Agent.NixFile.GitSource as GitSource import Hercules.Agent.NixPath ( renderSubPath, ) @@ -77,6 +82,7 @@ import qualified Hercules.Agent.WorkerProtocol.Event as Event import qualified Hercules.Agent.WorkerProtocol.Event.Attribute as WorkerAttribute import qualified Hercules.Agent.WorkerProtocol.Event.AttributeError as WorkerAttributeError +import qualified Hercules.Agent.WorkerProtocol.Event.AttributeIFD as AttributeIFD import qualified Hercules.Agent.WorkerProtocol.LogSettings as LogSettings import Hercules.CNix.Store (Store, StorePath, parseStorePath) import Hercules.Error (defaultRetry, quickRetry) @@ -89,6 +95,7 @@ import qualified System.Directory as Dir import System.FilePath import System.Process +import UnliftIO (atomicModifyIORef') eventLimit :: Int eventLimit = 50000 @@ -435,6 +442,7 @@ Eval.allowedPaths = allowedPaths } buildRequiredIndex <- liftIO $ newIORef (0 :: Int) + attributeIFDCounter <- liftIO $ newIORef (0 :: Int) commandChan <- newChan writeChan commandChan $ Just $ Command.Eval eval for_ (EvaluateTask.extraGitCredentials task) \creds -> @@ -470,7 +478,7 @@ ("protocol.file.allow", "always") ] } - withProducer (produceWorkerEvents eval envSettings commandChan) $ + withProducer (produceWorkerEvents (EvaluateTask.id task) eval envSettings commandChan) $ \workerEventsP -> fix $ \continue -> joinSTM $ listen @@ -501,6 +509,18 @@ AttributeErrorEvent.trace = WorkerAttributeError.trace e } continue + Event.AttributeIFD e -> do + index <- atomicModifyIORef' attributeIFDCounter \n -> (n + 1, n) + emit $ + EvaluateEvent.AttributeIFD $ + AttributeIFDEvent.AttributeIFDEvent + { AttributeIFDEvent.expressionPath = decode <$> AttributeIFD.path e, + AttributeIFDEvent.derivationPath = decode $ AttributeIFD.derivationPath e, + AttributeIFDEvent.derivationOutput = decode $ AttributeIFD.derivationOutput e, + AttributeIFDEvent.done = AttributeIFD.done e, + AttributeIFDEvent.index = index + } + continue Event.EvaluationDone -> writeChan commandChan Nothing Event.Error e -> do @@ -539,11 +559,17 @@ { derivationPath = drvText, forceRebuild = isJust notAttempt } - when waitForStatus do - flush - status <- drvPoller notAttempt drvText - logLocM DebugS $ "Got derivation status " <> logStr (show status :: Text) - writeChan commandChan $ Just $ Command.BuildResult $ uncurry (BuildResult.BuildResult drvText) status + let doPoll = do + status <- drvPoller notAttempt drvText + logLocM DebugS $ "Got derivation status " <> logStr (show status :: Text) + writeChan commandChan $ Just $ Command.BuildResult $ uncurry (BuildResult.BuildResult drvText) status + if waitForStatus + then do + flush + doPoll + else void $ do + uio <- askUnliftIO + liftIO $ forkIO $ unliftIO uio doPoll continue Event.OnPushHandler (ViaJSON e) -> do emit $ EvaluateEvent.OnPushHandlerEvent e @@ -580,12 +606,13 @@ pure $ toS $ Network.URI.uriRegName a produceWorkerEvents :: + Id (Task a) -> Eval.Eval -> WorkerProcess.WorkerEnvSettings -> Chan (Maybe Command.Command) -> (Event.Event -> App ()) -> App ExitCode -produceWorkerEvents eval envSettings commandChan writeEvent = do +produceWorkerEvents taskId eval envSettings commandChan writeEvent = do workerExe <- WorkerProcess.getWorkerExe let opts = [show $ Eval.extraNixOptions eval] -- NiceToHave: replace renderNixPath by something structured like -I @@ -597,7 +624,15 @@ close_fds = True, -- Disable on Windows? cwd = Just (Eval.cwd eval) } - WorkerProcess.runWorker wps (stderrLineHandler "evaluator") commandChan writeEvent + stderrHandler = + stderrLineHandler + ( M.fromList + [ ("taskId", A.toJSON taskId), + ("evalRev", A.toJSON (eval & Eval.gitSource & Event.fromViaJSON & GitSource.rev)) + ] + ) + "Effect worker" + WorkerProcess.runWorker wps stderrHandler commandChan writeEvent drvPoller :: Maybe UUID -> Text -> App (UUID, BuildResult.BuildStatus) drvPoller notAttempt drvPath = do diff --git a/hercules-ci-agent/Hercules/Agent/Init.hs b/hercules-ci-agent/Hercules/Agent/Init.hs --- a/hercules-ci-agent/Hercules/Agent/Init.hs +++ b/hercules-ci-agent/Hercules/Agent/Init.hs @@ -59,7 +59,7 @@ handleScribe <- K.mkHandleScribe K.ColorIfTerminal stderr (Compat.katipLevel (Config.logLevel cfg)) K.V2 let mkLogEnv = K.registerScribe "stderr" handleScribe K.defaultScribeSettings - =<< K.initLogEnv emptyNamespace "" + =<< K.initLogEnv (K.Namespace ["Agent"]) "" bracket mkLogEnv K.closeScribes f emptyNamespace :: K.Namespace diff --git a/hercules-ci-agent/Hercules/Agent/Log.hs b/hercules-ci-agent/Hercules/Agent/Log.hs --- a/hercules-ci-agent/Hercules/Agent/Log.hs +++ b/hercules-ci-agent/Hercules/Agent/Log.hs @@ -31,13 +31,15 @@ logLocM ErrorS $ logStr msg panic msg -stderrLineHandler :: KatipContext m => Text -> Int -> ByteString -> m () -stderrLineHandler _processRole _ ln +stderrLineHandler :: KatipContext m => Map Text Value -> Text -> Int -> ByteString -> m () +stderrLineHandler callerContext _processRole _ ln | "@katip " `BS.isPrefixOf` ln, Just item <- A.decode (LBS.fromStrict $ BS.drop 7 ln) = -- "This is the lowest level function [...] useful when implementing centralised logging services." - Katip.Core.logKatipItem (Katip.Core.SimpleLogPayload . M.toList . fmap (Katip.Core.AnyLogPayload :: A.Value -> Katip.Core.AnyLogPayload) <$> item) -stderrLineHandler processRole pid ln = + Katip.Core.logKatipItem (Katip.Core.SimpleLogPayload . M.toList . fmap (Katip.Core.AnyLogPayload :: A.Value -> Katip.Core.AnyLogPayload) . extendContext <$> item) + where + extendContext workerItem = M.union workerItem callerContext +stderrLineHandler _ processRole pid ln = withNamedContext "worker" (pid :: Int) $ logLocM InfoS $ logStr $ diff --git a/src/Hercules/Agent/Socket.hs b/src/Hercules/Agent/Socket.hs --- a/src/Hercules/Agent/Socket.hs +++ b/src/Hercules/Agent/Socket.hs @@ -1,4 +1,5 @@ {-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -32,7 +33,7 @@ import Network.WebSockets (Connection, runClientWith) import qualified Network.WebSockets as WS import Protolude hiding (atomically, handle, race, race_) -import UnliftIO.Async (race, race_) +import UnliftIO.Async (AsyncCancelled (AsyncCancelled), race, race_) import UnliftIO.Exception (handle) import UnliftIO.STM (readTVarIO) import UnliftIO.Timeout (timeout) @@ -133,7 +134,7 @@ liftIO (A.eitherDecode <$> WS.receiveData conn) >>= \case Left e -> liftIO $ throwIO (FatalError $ "Error decoding service message: " <> toS e) Right r -> pure r - handshake conn = katipAddNamespace "Handshake" do + handshake conn (removeHandshakeTimeout :: IO ()) = katipAddNamespace "Handshake" do siMsg <- recv conn case siMsg of Frame.Oob {o = o'} -> @@ -149,6 +150,7 @@ case ackMsg of Frame.Ack {n = n} -> cleanAcknowledged n _ -> throwIO $ FatalError "Expected acknowledgement. This is either a bug or you might need to update your agent." + liftIO removeHandshakeTimeout sendUnacked conn sendUnacked :: Connection -> m () sendUnacked conn = do @@ -222,13 +224,32 @@ -- terminate other threads via race_ pass else noAckCleanupThread' expectedN - forever $ + forever do + removeTimeout <- prepareTimeout handshakeTimeoutMicroseconds HandshakeTimeout handle logWarningPause $ withConnection' socketConfig $ \conn -> do katipAddNamespace "Handshake" do - handshake conn + handshake conn removeTimeout readThread conn `race_` writeThread conn `race_` noAckCleanupThread + +handshakeTimeoutMicroseconds :: Int +handshakeTimeoutMicroseconds = 30_000_000 + +data HandshakeTimeout = HandshakeTimeout + deriving (Show, Exception) + +prepareTimeout :: (Exception e, MonadIO m) => Int -> e -> m (IO ()) +prepareTimeout delay exc = do + requestingThread <- liftIO myThreadId + tid <- liftIO $ forkIO do + do + threadDelay delay + throwTo requestingThread exc + `catch` \(_ :: AsyncCancelled) -> + -- Removal of the timeout is normal, so do nothing + pass + pure $ throwTo tid AsyncCancelled msgN :: Frame o a -> Maybe Integer msgN Frame.Msg {n = n} = Just n diff --git a/src/Hercules/Agent/WorkerProtocol/Event.hs b/src/Hercules/Agent/WorkerProtocol/Event.hs --- a/src/Hercules/Agent/WorkerProtocol/Event.hs +++ b/src/Hercules/Agent/WorkerProtocol/Event.hs @@ -10,6 +10,7 @@ import Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent (OnPushHandlerEvent) import Hercules.Agent.WorkerProtocol.Event.Attribute import Hercules.Agent.WorkerProtocol.Event.AttributeError +import Hercules.Agent.WorkerProtocol.Event.AttributeIFD (AttributeIFD) import Hercules.Agent.WorkerProtocol.Event.BuildResult import Protolude hiding (get, put) import Prelude () @@ -17,6 +18,7 @@ data Event = Attribute Attribute | AttributeError AttributeError + | AttributeIFD AttributeIFD | EvaluationDone | Error Text | Build ByteString Text (Maybe UUID) Bool diff --git a/src/Hercules/Agent/WorkerProtocol/Event/AttributeIFD.hs b/src/Hercules/Agent/WorkerProtocol/Event/AttributeIFD.hs new file mode 100644 --- /dev/null +++ b/src/Hercules/Agent/WorkerProtocol/Event/AttributeIFD.hs @@ -0,0 +1,15 @@ +{-# LANGUAGE DeriveAnyClass #-} + +module Hercules.Agent.WorkerProtocol.Event.AttributeIFD where + +import Data.Binary (Binary) +import Protolude +import Prelude () + +data AttributeIFD = AttributeIFD + { path :: [ByteString], + derivationPath :: ByteString, + derivationOutput :: ByteString, + done :: Bool + } + deriving (Generic, Binary, Show, Eq) diff --git a/test/Hercules/Agent/Worker/ConduitSpec.hs b/test/Hercules/Agent/Worker/ConduitSpec.hs new file mode 100644 --- /dev/null +++ b/test/Hercules/Agent/Worker/ConduitSpec.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE BlockArguments #-} + +module Hercules.Agent.Worker.ConduitSpec where + +import Data.Conduit +import Data.Conduit.Combinators (sinkList) +import Data.Conduit.List (sourceList) +import Hercules.Agent.Worker.Conduit (tailC, takeCWhileStopEarly, withMessageLimit) +import Protolude hiding (yield) +import Test.Hspec + +spec :: Spec +spec = do + describe "tailC" do + it "can produce an empty output" do + l <- runConduit (pass .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [] + it "can produce a short output (1)" do + l <- runConduit (sourceList [1] .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [1] + it "can produce a short output (2)" do + l <- runConduit (sourceList [1, 2] .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [1, 2] + it "can produce a matching output (3)" do + l <- runConduit (sourceList [1 .. 3] .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [1, 2, 3] + it "can produce a tail output (4)" do + l <- runConduit (sourceList [1 .. 4] .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [2, 3, 4] + it "can produce a tail output (5)" do + l <- runConduit (sourceList [1 .. 5] .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [3, 4, 5] + it "can produce a tail output (100)" do + l <- runConduit (sourceList [1 .. 100] .| tailC 3 .| sinkList) + (l :: [Int]) `shouldBe` [98, 99, 100] + describe "takeCWhileStopEarly" do + it "works for example" do + (i, l) <- runConduit (sourceList [1 .. 10] .| (takeCWhileStopEarly even 2 `fuseBoth` sinkList)) + (l :: [Int]) `shouldBe` [1, 2, 3, 4] + i `shouldBe` (2, 4) + describe "withMessageLimit" do + let exampleConduit = do + withMessageLimit (const True) 20 10 (yield (-42)) (\between -> yield (-1 * between)) yield + + it "works for input 0" do + r <- runConduit do + sourceList ([] :: [Int]) + .| exampleConduit + .| sinkList + r `shouldBe` [] + + it "works for input 1" do + r <- runConduit do + sourceList [1 .. 12 :: Int] + .| exampleConduit + .| sinkList + r `shouldBe` [1 .. 12] + + it "works for input 2" do + r <- runConduit do + sourceList [1 .. 21 :: Int] + .| exampleConduit + .| sinkList + r `shouldBe` [1 .. 20] <> [-42, 21] + + it "works for input 3" do + r <- runConduit do + sourceList [1 .. 30 :: Int] + .| exampleConduit + .| sinkList + r `shouldBe` [1 .. 20] <> [-42] <> [21 .. 30] + + it "works for input 4" do + r <- runConduit do + sourceList [1 .. 31 :: Int] + .| exampleConduit + .| sinkList + r `shouldBe` [1 .. 20] <> [-42, -1] <> [22 .. 31] <> [1] + + it "works for input 5" do + r <- runConduit do + sourceList [1 .. 100 :: Int] + .| exampleConduit + .| sinkList + r `shouldBe` [1 .. 20] <> [-42, -70] <> [91 .. 100] <> [70] diff --git a/test/Hercules/Agent/Worker/STMSpec.hs b/test/Hercules/Agent/Worker/STMSpec.hs new file mode 100644 --- /dev/null +++ b/test/Hercules/Agent/Worker/STMSpec.hs @@ -0,0 +1,43 @@ +{-# LANGUAGE BlockArguments #-} + +module Hercules.Agent.Worker.STMSpec (spec) where + +import Control.Concurrent.STM (newEmptyTMVarIO, putTMVar, readTMVar) +import Hercules.Agent.Worker.STM +import Protolude +import Test.Hspec + +spec :: Spec +spec = do + describe "asyncIfSTM" do + it "can create a valid async" do + fooAsync <- + asyncIfSTM + (\_ -> pure (Right "world")) + \s -> do + pure ("hi, " <> s) + hi <- wait fooAsync + hi `shouldBe` ("hi, world" :: Text) + + it "can reuse an async" do + withAsync (pure "user") \firstAsync -> do + fooAsync <- + asyncIfSTM + (\_ -> pure (Left firstAsync)) + (panic "do not use") + hi <- wait fooAsync + hi `shouldBe` ("user" :: Text) + + it "can write the new async in the STM transaction" do + asyncHolder <- newEmptyTMVarIO + _irrelevant <- + asyncIfSTM + ( \asy -> do + putTMVar asyncHolder asy + pure (Right "user") + ) + \s -> do + pure ("hi, " <> s) + asy <- atomically $ readTMVar asyncHolder + hi <- wait asy + hi `shouldBe` ("hi, user" :: Text) diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -5,6 +5,8 @@ import qualified Hercules.Agent.Nix.RetrieveDerivationInfoSpec import qualified Hercules.Agent.NixPathSpec +import qualified Hercules.Agent.Worker.ConduitSpec +import qualified Hercules.Agent.Worker.STMSpec import qualified Hercules.Agent.WorkerProcessSpec import qualified Hercules.SecretsSpec import Test.Hspec @@ -15,3 +17,5 @@ describe "Hercules.Agent.WorkerProcessSpec" Hercules.Agent.WorkerProcessSpec.spec describe "Hercules.Agent.Nix.RetrieveDerivationInfo" Hercules.Agent.Nix.RetrieveDerivationInfoSpec.spec describe "Hercules.Secret" Hercules.SecretsSpec.spec + describe "Hercules.Agent.Worker.STMSpec" Hercules.Agent.Worker.STMSpec.spec + describe "Hercules.Agent.Worker.Conduit" Hercules.Agent.Worker.ConduitSpec.spec