hercules-ci-agent 0.9.13 → 0.10.1

raw patch · 60 files changed

+2637/−1145 lines, 60 filesdep +HUnitdep +QuickCheckdep +bifunctorsdep ~hercules-ci-api-agentdep ~hercules-ci-api-coredep ~tomlandPVP ok

version bump matches the API change (PVP)

Dependencies added: HUnit, QuickCheck, bifunctors, profunctors, tls

Dependency ranges changed: hercules-ci-api-agent, hercules-ci-api-core, tomland

API changes (from Hackage documentation)

Files

• CHANGELOG.md +40/−5
• cbits/hercules-aliases.h +0/−3
• cbits/hercules-logger.cxx +0/−11
• cbits/hercules-logger.hh +0/−5
• cbits/nix-2.4/hercules-store.cxx +35/−0
• cbits/nix-2.4/hercules-store.hh +32/−0
• hercules-ci-agent-worker/Hercules/Agent/Worker.hs +135/−284
• hercules-ci-agent-worker/Hercules/Agent/Worker/Build.hs +11/−18
• hercules-ci-agent-worker/Hercules/Agent/Worker/Build/Logger.hs +73/−103
• hercules-ci-agent-worker/Hercules/Agent/Worker/Build/Prefetched.hs +6/−0
• hercules-ci-agent-worker/Hercules/Agent/Worker/Conduit.hs +0/−78
• hercules-ci-agent-worker/Hercules/Agent/Worker/Effect.hs +8/−3
• hercules-ci-agent-worker/Hercules/Agent/Worker/Env.hs +5/−2
• hercules-ci-agent-worker/Hercules/Agent/Worker/Evaluate.hs +28/−8
• hercules-ci-agent-worker/Hercules/Agent/Worker/Logging.hs +3/−3
• hercules-ci-agent.cabal +44/−26
• hercules-ci-agent/Hercules/Agent.hs +32/−14
• hercules-ci-agent/Hercules/Agent/Build.hs +38/−30
• hercules-ci-agent/Hercules/Agent/Cache.hs +7/−0
• hercules-ci-agent/Hercules/Agent/Cachix.hs +1/−0
• hercules-ci-agent/Hercules/Agent/Cachix/Init.hs +2/−1
• hercules-ci-agent/Hercules/Agent/Client.hs +0/−12
• hercules-ci-agent/Hercules/Agent/Config.hs +94/−81
• hercules-ci-agent/Hercules/Agent/Config/Combined.hs +48/−0
• hercules-ci-agent/Hercules/Agent/Config/Json.hs +283/−0
• hercules-ci-agent/Hercules/Agent/Config/Toml.hs +58/−0
• hercules-ci-agent/Hercules/Agent/Effect.hs +14/−21
• hercules-ci-agent/Hercules/Agent/Env.hs +12/−1
• hercules-ci-agent/Hercules/Agent/EnvironmentInfo.hs +8/−5
• hercules-ci-agent/Hercules/Agent/Evaluate.hs +404/−225
• hercules-ci-agent/Hercules/Agent/Init.hs +6/−1
• hercules-ci-agent/Hercules/Agent/InitWorkerConfig.hs +23/−0
• hercules-ci-agent/Hercules/Agent/Log.hs +10/−7
• hercules-ci-agent/Hercules/Agent/LogSocket.hs +298/−0
• hercules-ci-agent/Hercules/Agent/Memo.hs +124/−0
• hercules-ci-agent/Hercules/Agent/Options.hs +2/−2
• hercules-ci-agent/Hercules/Agent/ResourceLimiter.hs +23/−0
• hercules-ci-nix-daemon/daemon.cc +4/−0
• src/Data/Conduit/Extras.hs +1/−25
• src/Hercules/Agent/Conduit.hs +78/−0
• src/Hercules/Agent/NixFile.hs +1/−0
• src/Hercules/Agent/Sensitive.hs +5/−15
• src/Hercules/Agent/Socket.hs +2/−2
• src/Hercules/Agent/Store.hs +24/−0
• src/Hercules/Agent/WorkerProcess.hs +18/−5
• src/Hercules/Agent/WorkerProtocol/Command/Build.hs +2/−3
• src/Hercules/Agent/WorkerProtocol/Command/Effect.hs +3/−4
• src/Hercules/Agent/WorkerProtocol/Command/Eval.hs +1/−5
• src/Hercules/Agent/WorkerProtocol/Event.hs +3/−0
• src/Hercules/Agent/WorkerProtocol/Event/BuildResult.hs +1/−12
• src/Hercules/Agent/WorkerProtocol/LogSettings.hs +0/−15
• src/Hercules/Agent/WorkerProtocol/OutputInfo.hs +20/−0
• src/Hercules/Agent/WorkerProtocol/WorkerConfig.hs +38/−0
• src/Hercules/Effect.hs +137/−9
• src/Hercules/Effect/Container.hs +40/−16
• src/Hercules/Secrets.hs +3/−0
• test/Hercules/Agent/ConduitSpec.hs +85/−0
• test/Hercules/Agent/ConfigSpec.hs +236/−0
• test/Hercules/Agent/Worker/ConduitSpec.hs +0/−85
• test/Hercules/Agent/WorkerProtocol/WorkerConfigSpec.hs +28/−0

@@ -5,16 +5,50 @@ 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.13] - 2023-03-09+## [0.10.1] - 2024-02-12 +### Changed++ - More work is performed concurrently during evaluation, including binary cache lookups and (more) build dispatch. This results in a speedup.++ - Dependencies of build dependencies are not scheduled eagerly anymore.+   This reduces the scope of all jobs that are evaluated by agents since this release, resulting in a speedup.+   This resolves a noticable slowdown when first evaluating significant Nixpkgs updates when its `staging` branch is merged.++   Strictly speaking, a job success no longer guarantees that absolutely everything (all the way up to the bootstrap binaries) is realisable on your agents.+   This property is generally not your responsibility, and enforcing it had the effect of excluding less reproducible platforms such as darwin.+   Instead, a weaker property is provided: *your* derivations are realisable, as well as the immediate build dependencies. "Your derivations" is defined as those whose outputs are not already cached.++   CI setups based on the Nix command line interface (almost all CIs) also behave this way.++ - The recommended configuration format is now **JSON**, preferably generated using a configuration manager such as NixOS or nix-darwin.+   TOML is still supported, but does not support `null` in labels, and due to library limitations, it requires that intermediate tables be specified. See [the config file documentation](https://docs.hercules-ci.com/hercules-ci-agent/agent-config).++ - `services.hercules-ci-agent` is now an alias for `services.hercules-ci-agents.""`, which still provides the same behavior as the old module.++ - Hardening flags have been applied to the NixOS module.++ - The effect sandbox now use the `crun` container runtime instead of `runc`.++ - Attribute sets containing a `_type` attribute are not scanned for derivations in `herculesCI.<...>.outputs`. This prevents accidental scanning of large or failing attribute trees, such as NixOS configurations. `nixosConfigurations` in Flakes are still built as usual, as they are not (verbatim) in the `herculesCI.<...>.outputs` attributes.+ ### Added - - Nix 2.18 support+ - Effect mounts. Specify [`effectMountables`](https://docs.hercules-ci.com/hercules-ci-agent/agent-config.html#effectMountables) in the agent configuration, deploy, and [mount](https://docs.hercules-ci.com/hercules-ci-agent/effects/declaration.html#__hci_effect_mounts) them into an effect. This can be used for instance to expose the host's `/etc/hosts`, or hardware devices such as GPUs. Access is [controlled](https://docs.hercules-ci.com/hercules-ci-agent/agent-config.html#effectMountables-condition) by the agent configuration. - - Nix 2.17 support+ - New configuration option `remotePlatformsWithSameFeatures`, allowing a remote build to be used before more elaborate remote builder support is implemented.+   The recommended method for running a cluster is still to install `hercules-ci-agent` on each machine, as that is more efficient and accurate. -## [0.9.12] - 2022-06-28+ - Agent [labels](https://docs.hercules-ci.com/hercules-ci-agent/agent-config.html#labels) can now be `null`, when using the JSON configuration format. +### Fixed++ - Low level crash details are now reported in the log as expected.++ - An interaction between the Nix GC and threads has been fixed, solving such a crash.++## [0.9.12] - 2023-06-28+ ### Added   - Nix 2.16 support@@ -25,7 +59,7 @@   - Do not `chdir` the build worker. This functionality of the `process` package appears unreliable, but is not needed. -## [0.9.11] - 2022-03-06+## [0.9.11] - 2023-03-06  ### BREAKING @@ -724,6 +758,7 @@  - Initial release +[0.9.12]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.11...hercules-ci-agent-0.9.12 [0.9.11]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.10...hercules-ci-agent-0.9.11 [0.9.10]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.9...hercules-ci-agent-0.9.10 [0.9.9]: https://github.com/hercules-ci/hercules-ci-agent/compare/hercules-ci-agent-0.9.8...hercules-ci-agent-0.9.9

@@ -20,9 +20,6 @@  typedef nix::Strings::iterator StringsIterator; typedef nix::DerivationOutputs::iterator DerivationOutputsIterator;-#if ! MIN_VERSION_hercules_ci_cnix_store(0,3,5)-typedef nix::DerivationInputs::iterator DerivationInputsIterator;-#endif typedef nix::StringPairs::iterator StringPairsIterator;  using namespace std;

@@ -6,18 +6,11 @@   wakeup.notify_one(); } -uint64_t HerculesLogger::getMs() {-  auto t = std::chrono::steady_clock::now();-  auto millis = std::chrono::duration_cast<std::chrono::milliseconds>(t - t_zero).count();-  return millis;-}- #if NIX_IS_AT_LEAST(2, 15, 0) void HerculesLogger::log(nix::Verbosity lvl, std::string_view s) {   push(std::make_unique<LogEntry>(LogEntry {     .entryType = 1,     .level = lvl,-    .ms = getMs(),     .text = std::string(s)   })); }@@ -26,7 +19,6 @@   push(std::make_unique<LogEntry>(LogEntry {     .entryType = 1,     .level = lvl,-    .ms = getMs(),     .text = fs.s   })); }@@ -46,7 +38,6 @@   push(std::make_unique<LogEntry>(LogEntry {     .entryType = 2,     .level = lvl,-    .ms = getMs(),     .text = s,     .activityId = act,     .type = type,@@ -58,7 +49,6 @@ void HerculesLogger::stopActivity(nix::ActivityId act) {   push(std::make_unique<LogEntry>(LogEntry {     .entryType = 3,-    .ms = getMs(),     .activityId = act   })); }@@ -66,7 +56,6 @@ void HerculesLogger::result(nix::ActivityId act, nix::ResultType type, const Fields & fields) {   push(std::make_unique<LogEntry>(LogEntry {     .entryType = 4,-    .ms = getMs(),     .activityId = act,     .type = type,     .fields = fields

@@ -7,7 +7,6 @@ #include <nix/logging.hh> #include <queue> #include <string>-#include <chrono>  class HerculesLogger final : public nix::Logger { @@ -16,8 +15,6 @@  private: -  std::chrono::time_point<std::chrono::steady_clock> t_zero = std::chrono::steady_clock::now();-   struct State {     std::queue<std::unique_ptr<LogEntry>> queue;   };@@ -25,7 +22,6 @@   std::condition_variable wakeup;    void push(std::unique_ptr<LogEntry> entry);-  uint64_t getMs();   public:   inline HerculesLogger() {};@@ -33,7 +29,6 @@   struct LogEntry {     int entryType;     int level;-    uint64_t ms;     std::string text;     uint64_t activityId;     uint64_t type;

@@ -84,7 +84,35 @@   wrappedStore->addToStore(info, narSource, repair, checkSigs); } +#if NIX_IS_AT_LEAST(2,20,0)+ StorePath WrappingStore::addToStore(+    std::string_view name,+    SourceAccessor & accessor,+    const CanonPath & path,+    ContentAddressMethod method,+    HashAlgorithm hashAlgo,+    const StorePathSet & references,+    PathFilter & filter,+    RepairFlag repair)+{+    return wrappedStore->addToStore(name, accessor, path, method, hashAlgo, references, filter, repair);+}++StorePath WrappingStore::addToStoreFromDump(+    Source & dump,+    std::string_view name,+    ContentAddressMethod method,+    HashAlgorithm hashAlgo,+    const StorePathSet & references,+    RepairFlag repair)+{+    return wrappedStore->addToStoreFromDump(dump, name, method, hashAlgo, references, repair);+}++#else // < 2.20++StorePath WrappingStore::addToStore( #if NIX_IS_AT_LEAST(2,7,0)       std::string_view name, #else@@ -137,6 +165,7 @@       const StorePathSet & references, RepairFlag repair) {   return wrappedStore->addTextToStore(name, s, references, repair); }+#endif // < 2.20  void WrappingStore::narFromPath(const StorePath& path, Sink& sink) {   wrappedStore->narFromPath(path, sink);@@ -191,9 +220,15 @@   return wrappedStore->verifyStore(checkContents, repair); }; +#if NIX_IS_AT_LEAST(2,19,0)+ref<FSAccessor> WrappingStore::getFSAccessor(bool requireValidPath) {+  return wrappedStore->getFSAccessor(requireValidPath);+}+#else ref<FSAccessor> WrappingStore::getFSAccessor() {   return wrappedStore->getFSAccessor(); }+#endif  void WrappingStore::addSignatures(const StorePath& storePath,                                   const StringSet& sigs) {

@@ -14,6 +14,10 @@ #endif #include "HsFFI.h" +#if NIX_IS_AT_LEAST(2,19,0)+using FSAccessor = nix::SourceAccessor;+#endif+ using namespace nix;  class WrappingStore : public Store {@@ -60,6 +64,28 @@   virtual void addToStore(const ValidPathInfo & info, Source & narSource,       RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) override; +#if NIX_IS_AT_LEAST(2,20,0)++    virtual StorePath addToStore(+        std::string_view name,+        SourceAccessor & accessor,+        const CanonPath & path,+        ContentAddressMethod method = FileIngestionMethod::Recursive,+        HashAlgorithm hashAlgo = HashAlgorithm::SHA256,+        const StorePathSet & references = StorePathSet(),+        PathFilter & filter = defaultPathFilter,+        RepairFlag repair = NoRepair) override;++    virtual StorePath addToStoreFromDump(+        Source & dump,+        std::string_view name,+        ContentAddressMethod method = FileIngestionMethod::Recursive,+        HashAlgorithm hashAlgo = HashAlgorithm::SHA256,+        const StorePathSet & references = StorePathSet(),+        RepairFlag repair = NoRepair) override;+++#else // < 2.20   virtual StorePath addToStore( #if NIX_IS_AT_LEAST(2,7,0)       std::string_view name,@@ -97,6 +123,8 @@ #endif     const StorePathSet & references, RepairFlag repair = NoRepair) override; +#endif // < 2.20+   virtual void narFromPath(const StorePath & path, Sink & sink) override;    virtual void buildPaths(@@ -127,7 +155,11 @@    virtual bool verifyStore(bool checkContents, RepairFlag repair = NoRepair) override; +#if NIX_IS_AT_LEAST(2,19,0)+  virtual ref<FSAccessor> getFSAccessor(bool requireValidPath) override;+#else   virtual ref<FSAccessor> getFSAccessor() override;+#endif    virtual void addSignatures(const StorePath & storePath, const StringSet & sigs) override; 

@@ -11,33 +11,27 @@  import Conduit import Control.Concurrent.STM hiding (check)-import Control.Exception.Lifted qualified as EL import Control.Monad.Except import Control.Monad.IO.Unlift+import Data.Aeson qualified as A+import Data.Binary qualified as B+import Data.ByteString.Char8 qualified as BSC+import Data.ByteString.Lazy qualified as BL import Data.Conduit qualified-import Data.Conduit.Extras (sinkChan, sinkChanTerminate, sourceChan) import Data.Conduit.Katip.Orphans () import Data.Conduit.Serialization.Binary   ( conduitDecode,-    conduitEncode,   ) import Data.IORef import Data.Map qualified as M import Data.Vector (Vector) import Data.Vector qualified as V-import Hercules.API.Agent.LifeCycle.ServiceInfo qualified-import Hercules.API.Logs.LogEntry (LogEntry)-import Hercules.API.Logs.LogEntry qualified as LogEntry-import Hercules.API.Logs.LogHello (LogHello (LogHello, clientProtocolVersion, storeProtocolVersion))-import Hercules.API.Logs.LogMessage (LogMessage)-import Hercules.API.Logs.LogMessage qualified as LogMessage-import Hercules.Agent.Sensitive-import Hercules.Agent.Socket qualified as Socket+import GHC.Natural (Natural) import Hercules.Agent.Worker.Build (runBuild) import Hercules.Agent.Worker.Build.Logger qualified as Logger-import Hercules.Agent.Worker.Conduit (takeCWhileStopEarly, withMessageLimit) import Hercules.Agent.Worker.Effect (runEffect)-import Hercules.Agent.Worker.Env (HerculesState (..))+import Hercules.Agent.Worker.Env (HerculesState (HerculesState, drvsCompleted, wrappedStore))+import Hercules.Agent.Worker.Env qualified import Hercules.Agent.Worker.Error (ExceptionText (exceptionTextMessage), exceptionTextMessage, renderException) import Hercules.Agent.Worker.Evaluate (runEval) import Hercules.Agent.Worker.HerculesStore (setBuilderCallback, withHerculesStore)@@ -46,9 +40,7 @@   ( Command,   ) import Hercules.Agent.WorkerProtocol.Command qualified as Command-import Hercules.Agent.WorkerProtocol.Command.Build qualified as Build import Hercules.Agent.WorkerProtocol.Command.BuildResult qualified as BuildResult-import Hercules.Agent.WorkerProtocol.Command.Effect qualified as Effect import Hercules.Agent.WorkerProtocol.Command.Eval   ( Eval,   )@@ -57,66 +49,108 @@   ( Event (Exception),   ) import Hercules.Agent.WorkerProtocol.Event qualified as Event-import Hercules.Agent.WorkerProtocol.LogSettings qualified as LogSettings+import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON (ViaJSON))+import Hercules.Agent.WorkerProtocol.WorkerConfig (WorkerConfig)+import Hercules.Agent.WorkerProtocol.WorkerConfig qualified import Hercules.CNix as CNix-import Hercules.CNix.Expr (init, setExtraStackOverflowHandlerToSleep)+import Hercules.CNix.Expr (allowThreads, init, runGcRegisteredThread, setExtraStackOverflowHandlerToSleep) import Hercules.CNix.Util (installDefaultSigINTHandler)-import Hercules.CNix.Verbosity (setShowTrace)+import Hercules.CNix.Verbosity (setShowTrace, setVerbosity)+import Hercules.CNix.Verbosity qualified as CNix.Verbosity import Hercules.Error import Katip-import Network.URI qualified import Protolude hiding (bracket, catch, check, evalState, wait, withAsync, yield)-import System.Environment qualified as Environment-import System.IO (BufferMode (LineBuffering), hSetBuffering)+import System.IO (BufferMode (NoBuffering), hFlush, hSetBuffering) import System.Posix.Signals (Handler (Catch), installHandler, raiseSignal, sigINT, sigTERM)-import System.Timeout (timeout) import UnliftIO.Async (wait, withAsync)-import Prelude ()-import Prelude qualified  main :: IO () main = do-  hSetBuffering stderr LineBuffering-  Hercules.CNix.Expr.init-  setShowTrace True-  _ <- installHandler sigTERM (Catch $ raiseSignal sigINT) Nothing-  installDefaultSigINTHandler+  parentHandles <- Logger.plumbWorkerStd+  withEventWriter parentHandles.events \sendEvents -> do+    withNixLogger sendEvents do+      Hercules.CNix.Expr.init+      Hercules.CNix.Expr.allowThreads+      setShowTrace True+      _ <- installHandler sigTERM (Catch $ raiseSignal sigINT) Nothing+      installDefaultSigINTHandler+      setExtraStackOverflowHandlerToSleep+      workerConfigBytes <- BSC.hGetLine parentHandles.commands+      workerConfig <- case A.eitherDecode (toS workerConfigBytes) of+        Left e -> panic ("Could not parse WorkerConfig: " <> show e)+        Right r -> pure r+      taskWorker parentHandles.commands sendEvents workerConfig++withNixLogger ::+  (Vector Event -> IO ()) ->+  IO r ->+  IO r+withNixLogger sendEvents work = do   Logger.initLogger-  setExtraStackOverflowHandlerToSleep-  args <- Environment.getArgs-  case args of-    [options] -> taskWorker options-    _ -> throwIO $ FatalError "worker: Unrecognized command line arguments"+  Logger.withLoggerConduit logger work+  where+    logger c =+      runConduit $+        c .| awaitForever \events -> do+          liftIO . sendEvents . pure . Event.LogItems . ViaJSON $ events --- TODO Make this part of the worker protocol instead-parseOptions :: (Read a, Read b, IsString a, IsString b) => Prelude.String -> [(a, b)]-parseOptions options =-  Prelude.read options-    ++ [-         -- narinfo-cache-negative-ttl: Always try requesting narinfos because it may have been built in the meanwhile-         ("narinfo-cache-negative-ttl", "0"),-         -- Build concurrency is controlled by hercules-ci-agent, so set it-         -- to 1 to avoid accidentally consuming too many resources at once.-         ("max-jobs", "1")-       ]+-- | Run a pipe. Wait for producer to complete and then wait for consumer to exit.+withProducerDrivenBoundedQueue ::+  -- Max queue size+  Natural ->+  (forall x. TBQueue (Either x b) -> IO r) ->+  (forall x. TBQueue (Either x b) -> IO x) ->+  IO r+withProducerDrivenBoundedQueue maxQueueSize producer consumer = do+  q <- newTBQueueIO maxQueueSize+  let produce = do+        ( do+            r <- producer q+            atomically do+              writeTBQueue q (Left r)+          )+          `onException` do+            atomically do+              writeTBQueue q (Left (panic "producer threw an exception")) -setOptions :: [Char] -> IO ()-setOptions options = do-  for_ (parseOptions options) $ \(k, v) -> do-    setGlobalOption k v-    setOption k v+      consume = \producerDone -> do+        consumer q `finally` wait producerDone+  withAsync produce consume -taskWorker :: [Char] -> IO ()-taskWorker options = do-  setOptions options+withEventWriter :: Handle -> ((Vector Event -> IO ()) -> IO r) -> IO r+withEventWriter h f = do+  hSetBuffering h NoBuffering+  withProducerDrivenBoundedQueue 100 source shovel+  where+    source q = f (atomically . writeTBQueue q . Right)+    shovel q = do+      atomically (readTBQueue q) >>= \case+        Left r -> pure r+        Right events -> do+          unless (V.null events) do+            for_ events (BL.hPut h . B.encode)+            hFlush h+          shovel q++setOptions :: [(Text, Text)] -> IO ()+setOptions = traverse_ \(k, v) -> do+  setGlobalOption k v+  setOption k v++taskWorker :: Handle -> (Vector Event -> IO ()) -> WorkerConfig -> IO ()+taskWorker commandsHandle sendEvents_ cfg = do+  setOptions cfg.nixOptions+  -- at least info messages are expected in the dashboard logs+  let nixVerbosity = CNix.Verbosity.both CNix.Verbosity.Info cfg.nixVerbosity.unViaShowRead+      withKatip' = withKatip cfg.verbosity.unViaShowRead+  setVerbosity nixVerbosity   drvsCompleted_ <- newTVarIO mempty   drvBuildAsyncs_ <- newTVarIO mempty   drvRebuildAsyncs_ <- newTVarIO mempty   drvOutputSubstituteAsyncs_ <- newTVarIO mempty   drvsInProgress_ <- newIORef mempty-  withStore $ \wrappedStore_ -> withHerculesStore wrappedStore_ $ \herculesStore_ -> withKatip $ do+  withStore $ \wrappedStore_ -> withHerculesStore wrappedStore_ $ \herculesStore_ -> withKatip' do     liftIO $ setBuilderCallback herculesStore_ mempty-    ch <- liftIO newChan     let st =           HerculesState             { drvsCompleted = drvsCompleted_,@@ -126,46 +160,26 @@               drvsInProgress = drvsInProgress_,               herculesStore = herculesStore_,               wrappedStore = wrappedStore_,-              shortcutChannel = ch,-              extraNixOptions = parseOptions options+              sendEvents = sendEvents_,+              extraNixOptions = cfg.nixOptions,+              nixVerbosity = cfg.nixVerbosity.unViaShowRead             }-    let runner :: KatipContextT IO ()-        runner =-          ( ( do-                command <--                  runConduitRes -- Res shouldn't be necessary-                    ( transPipe liftIO (sourceHandle stdin)-                        .| conduitDecode-                        .| printCommands-                        .| await-                    )-                    >>= \case-                      Just c -> pure c-                      Nothing -> panic "Not a valid starting command"-                runCommand st ch command-            )-              `safeLiftedCatch` ( \e -> liftIO $ do-                                    textual <- renderException e-                                    writeChan ch (Just $ Exception $ exceptionTextMessage textual)-                                    exitFailure-                                )-          )-            `EL.finally` ( do-                             liftIO $ writeChan ch Nothing-                             logLocM DebugS "runner done"-                         )-        writer =-          runConduitRes-            ( sourceChan ch-                .| conduitEncode-                .| concatMapC (\x -> [Chunk x, Flush])-                .| transPipe liftIO (sinkHandleFlush stdout)-            )-    void $-      withAsync runner $ \runnerAsync -> do-        writer -- runner can stop writer only by passing Nothing in channel (finally)-        logLocM DebugS "Writer done"-        wait runnerAsync -- include the potential exception+    runConduitRes do+      sourceHandle commandsHandle+        .| conduitDecode+        .| printCommands+        .| do+          command <-+            await >>= \case+              Just c -> pure c+              Nothing -> panic "No starting command"+          runCommand st command+        `catchC` ( \e -> do+                     textual <- liftIO (renderException e)+                     yield (Exception $ exceptionTextMessage textual)+                     liftIO exitFailure+                 )+        .| awaitForever \x -> liftIO do sendEvents_ (pure x)  printCommands :: (KatipContext m) => ConduitT Command Command m () printCommands =@@ -176,38 +190,21 @@         pure x     ) -connectCommand ::-  (MonadUnliftIO m, KatipContext m, MonadThrow m) =>-  Chan (Maybe Event) ->-  ConduitM Command Event (ResourceT m) () ->-  m ()-connectCommand ch conduit =-  runConduitRes-    ( sourceHandle stdin-        .| conduitDecode-        .| printCommands-        .| conduit-        .| sinkChan ch-    )--runCommand :: (MonadUnliftIO m, KatipContext m, MonadThrow m) => HerculesState -> Chan (Maybe Event) -> Command -> m ()--- runCommand' :: HerculesState -> Command -> ConduitM Command Event (ResourceT IO) ()-runCommand herculesState ch command = do+runCommand :: (MonadUnliftIO m, KatipContext m, MonadThrow m) => HerculesState -> Command -> ConduitM Command Event m ()+runCommand herculesState command = do   -- TODO don't do this   mainThread <- liftIO myThreadId-  UnliftIO unlift <- askUnliftIO-  protocolVersion <- liftIO do-    getStoreProtocolVersion (wrappedStore herculesState)+   case command of-    Command.Eval eval -> Logger.withLoggerConduit (logger (Eval.logSettings eval) protocolVersion) $-      Logger.withTappedStderr Logger.tapper $-        connectCommand ch $ do-          liftIO $ restrictEval eval-          void-            $ liftIO-            $ flip-              forkFinally-              (escalateAs \e -> FatalError $ "Failed to fork: " <> show e)+    Command.Eval eval -> do+      liftIO $ restrictEval eval+      void $ lift do+        UnliftIO unlift <- askUnliftIO+        liftIO do+          flip+            forkFinally+            (escalateAs \e -> FatalError $ "Failed to fork: " <> show e)+            $ runGcRegisteredThread             $ unlift             $ runConduitRes               ( Data.Conduit.handleC@@ -219,31 +216,25 @@                       runEval herculesState eval                       liftIO $ throwTo mainThread ExitSuccess                   )-                  .| sinkChanTerminate (shortcutChannel herculesState)+                  .| awaitForever (liftIO . herculesState.sendEvents . pure)               )-          awaitForever $ \case-            Command.BuildResult (BuildResult.BuildResult path attempt result) -> do-              katipAddContext (sl "path" path <> sl "result" (show result :: Text)) $-                logLocM-                  DebugS-                  "Received remote build result"-              storePath <- liftIO $ CNix.parseStorePath (wrappedStore herculesState) (encodeUtf8 path)-              liftIO $ atomically $ modifyTVar (drvsCompleted herculesState) (M.insert storePath (attempt, result))-            _ -> pass+      awaitForever $ \case+        Command.BuildResult (BuildResult.BuildResult path attempt result) -> do+          katipAddContext (sl "path" path <> sl "result" (show result :: Text)) $+            logLocM+              DebugS+              "Received remote build result"+          storePath <- liftIO $ CNix.parseStorePath (wrappedStore herculesState) (encodeUtf8 path)+          liftIO $ atomically $ modifyTVar (drvsCompleted herculesState) (M.insert storePath (attempt, result))+        _ -> pass     Command.Build build ->       katipAddNamespace "Build" $-        Logger.withLoggerConduit (logger (Build.logSettings build) protocolVersion) $-          Logger.withTappedStderr Logger.tapper $-            connectCommand ch $-              runBuild (wrappedStore herculesState) build+        runBuild (wrappedStore herculesState) build     Command.Effect effect ->       katipAddNamespace "Effect" $-        Logger.withLoggerConduit (logger (Effect.logSettings effect) protocolVersion) $-          Logger.withTappedStderr Logger.tapper $-            connectCommand ch $ do-              runEffect (extraNixOptions herculesState) (wrappedStore herculesState) effect >>= \case-                ExitSuccess -> yield $ Event.EffectResult 0-                ExitFailure n -> yield $ Event.EffectResult n+        runEffect (herculesState.extraNixOptions) (wrappedStore herculesState) effect >>= \case+          ExitSuccess -> yield $ Event.EffectResult 0+          ExitFailure n -> yield $ Event.EffectResult n     _ ->       panic "Not a valid starting command" @@ -255,145 +246,5 @@       then allSchemes       else safeSchemes   where-    safeSchemes = "ssh:// https://"+    safeSchemes = "ssh:// https:// git+ssh:// git+https:// github: gitlab: sourcehut:"     allSchemes = safeSchemes <> " http:// git://"--logger :: (MonadUnliftIO m, KatipContext m) => LogSettings.LogSettings -> Int -> ConduitM () (Vector LogEntry) m () -> m ()-logger logSettings_ storeProtocolVersionValue entriesSource = do-  socketConfig <- liftIO $ makeSocketConfig logSettings_ storeProtocolVersionValue-  let withPings socket m =-        withAsync-          ( liftIO $ forever do-              -- TODO add ping constructor to Frame or use websocket pings-              let ping = LogMessage.LogEntries mempty-              threadDelay 30_000_000-              atomically $ Socket.write socket ping-          )-          (const m)-  Socket.withReliableSocket socketConfig $ \socket -> withPings socket do-    let conduit =-          entriesSource-            .| Logger.unbatch-            .| Logger.filterProgress-            .| dropMiddle-            .| renumber 0-            .| batchAndEnd-            .| socketSink socket-        batch = Logger.batch .| mapC (LogMessage.LogEntries . V.fromList)-        batchAndEnd =-          (foldMapTap (Last . ims) `fuseUpstream` batch) >>= \case-            Last (Just (i, ms)) -> yield $ LogMessage.End {i = i + 1, ms = ms}-            Last Nothing -> yield $ LogMessage.End 0 0-          where-            ims (Chunk logEntry) = Just (LogEntry.i logEntry, LogEntry.ms logEntry)-            ims _ = Nothing-        renumber i =-          await >>= traverse_ \case-            Flush -> yield Flush >> renumber i-            Chunk e -> do-              yield $ Chunk e {LogEntry.i = i}-              renumber (i + 1)-    runConduit conduit-    logLocM DebugS "Syncing"-    liftIO (timeout 600_000_000 $ Socket.syncIO socket) >>= \case-      Just _ -> pass-      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---- | Perform a foldMap while yielding the original values ("tap").------ '<>' is invoked with the new b on the right.-foldMapTap :: (Monoid b, Monad m) => (a -> b) -> ConduitT a a m b-foldMapTap f = go mempty-  where-    go b =-      await >>= \case-        Nothing -> pure b-        Just a -> do-          yield a-          go (b <> f a)--makeSocketConfig :: (MonadIO m) => LogSettings.LogSettings -> Int -> IO (Socket.SocketConfig LogMessage Hercules.API.Agent.LifeCycle.ServiceInfo.ServiceInfo m)-makeSocketConfig l storeProtocolVersionValue = do-  clientProtocolVersionValue <- liftIO getClientProtocolVersion-  baseURL <- case Network.URI.parseURI $ toS $ LogSettings.baseURL l of-    Just x -> pure x-    Nothing -> panic "LogSettings: invalid base url"-  pure-    Socket.SocketConfig-      { makeHello =-          pure $-            LogMessage.Hello-              LogHello-                { clientProtocolVersion = clientProtocolVersionValue,-                  storeProtocolVersion = storeProtocolVersionValue-                },-        checkVersion = Socket.checkVersion',-        baseURL = baseURL,-        path = LogSettings.path l,-        token = encodeUtf8 $ reveal $ LogSettings.token l-      }

@@ -5,6 +5,7 @@  import Conduit import Data.Conduit.Katip.Orphans ()+import Hercules.Agent.Store (toDrvInfo) import Hercules.Agent.Worker.Build.Prefetched (buildDerivation) import Hercules.Agent.Worker.Build.Prefetched qualified as Build import Hercules.Agent.WorkerProtocol.Command.Build qualified as Command.Build@@ -12,23 +13,24 @@ import Hercules.Agent.WorkerProtocol.Event qualified as Event import Hercules.Agent.WorkerProtocol.Event.BuildResult qualified as Event.BuildResult import Hercules.CNix-  ( DerivationOutput (derivationOutputName, derivationOutputPath),-    getDerivationOutputs,+  ( getDerivationOutputs,   ) import Hercules.CNix qualified as CNix-import Hercules.CNix.Store (Store, queryPathInfo, validPathInfoNarHash32, validPathInfoNarSize)+import Hercules.CNix.Store (Store) import Katip import Protolude hiding (yield)  runBuild :: (KatipContext m) => Store -> Command.Build.Build -> ConduitT i Event m ()-runBuild store build = do+runBuild store build = katipAddContext (sl "derivationPath" build.drvPath) do+  logLocM DebugS "runBuild"   let extraPaths = Command.Build.inputDerivationOutputPaths build       drvPath = encodeUtf8 $ Command.Build.drvPath build   drvStorePath <- liftIO $ CNix.parseStorePath store drvPath   x <- for extraPaths $ \input -> liftIO $ do     storePath <- CNix.parseStorePath store input+    CNix.addTemporaryRoot store storePath     try $ CNix.ensurePath store storePath-  materialize <- case sequenceA x of+  materialize0 <- case sequenceA x of     Right _ ->       -- no error, proceed with requested materialization setting       pure $ Command.Build.materializeDerivation build@@ -41,6 +43,9 @@   derivation <- case derivationMaybe of     Just drv -> pure drv     Nothing -> panic $ "Could not retrieve derivation " <> show drvStorePath <> " from local store or binary caches."+  drvPlatform <- liftIO $ CNix.getDerivationPlatform derivation+  let mayNeedRemoteBuild = drvPlatform `elem` Command.Build.materializePlatforms build+      materialize = materialize0 || mayNeedRemoteBuild   nixBuildResult <- liftIO $ buildDerivation store drvStorePath derivation (extraPaths <$ guard (not materialize))   katipAddContext (sl "result" (show nixBuildResult :: Text)) $     logLocM DebugS "Build result"@@ -54,17 +59,5 @@     Event.BuildResult.BuildFailure {errorMessage = Build.errorMessage result} enrichResult store drvName derivation _ = do   drvOuts <- getDerivationOutputs store drvName derivation-  outputInfos <- for drvOuts $ \drvOut -> do-    -- FIXME: ca-derivations: always get the built path-    vpi <- for (derivationOutputPath drvOut) (queryPathInfo store)-    hash_ <- traverse validPathInfoNarHash32 vpi-    path <- traverse (CNix.storePathToPath store) (derivationOutputPath drvOut)-    let size = fmap validPathInfoNarSize vpi-    pure-      Event.BuildResult.OutputInfo-        { name = derivationOutputName drvOut,-          path = fromMaybe "" path,-          hash = fromMaybe "" hash_,-          size = fromMaybe 0 size-        }+  outputInfos <- for drvOuts $ toDrvInfo store   pure $ Event.BuildResult.BuildSuccess outputInfos

@@ -3,28 +3,30 @@ {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE TemplateHaskell #-} -module Hercules.Agent.Worker.Build.Logger (initLogger, withLoggerConduit, tapper, withTappedStderr, batch, unbatch, filterProgress, nubProgress) where+module Hercules.Agent.Worker.Build.Logger+  ( initLogger,+    withLoggerConduit,+    nubProgress,+    plumbWorkerStd,+    Handles (..),+  )+where -import Conduit (MonadUnliftIO, filterC)-import Data.ByteString.Char8 qualified as BSC+import Conduit (MonadUnliftIO) import Data.ByteString.Unsafe (unsafePackMallocCString)-import Data.Conduit (ConduitT, Flush (..), await, awaitForever, yield)+import Data.Conduit (ConduitT, Flush (..), await, yield) import Data.Vector (Vector) import Data.Vector qualified as V import Foreign (alloca, nullPtr, peek)+import GHC.IO.Handle (hDuplicate, hDuplicateTo) import Hercules.API.Logs.LogEntry (LogEntry) import Hercules.API.Logs.LogEntry qualified as LogEntry import Hercules.Agent.Worker.Build.Logger.Context (Fields, HerculesLoggerEntry, context) import Hercules.CNix.Store.Context (unsafeMallocBS)-import Katip import Language.C.Inline.Cpp qualified as C import Language.C.Inline.Cpp.Exception qualified as C import Protolude hiding (bracket, finally, mask_, onException, tryJust, wait, withAsync, yield)-import System.IO (BufferMode (LineBuffering), hSetBuffering)-import System.IO.Error (isEOFError)-import System.Posix.IO (closeFd, createPipe, dup, dupTo, fdToHandle, stdError)-import System.Posix.Internals (setNonBlockingFD)-import System.Posix.Types (Fd)+import System.IO (BufferMode (LineBuffering, NoBuffering), hSetBuffering) import System.Timeout (timeout) import UnliftIO.Async import UnliftIO.Exception@@ -121,13 +123,15 @@ forNonNull p f = if p == nullPtr then pure Nothing else Just <$> f p  -- popping multiple lines into an array would be nice-convertEntry :: Ptr HerculesLoggerEntry -> IO LogEntry-convertEntry logEntryPtr = alloca \millisPtr -> alloca \textStrPtr -> alloca \levelPtr -> alloca \activityIdPtr -> alloca \typePtr -> alloca \parentPtr -> alloca \fieldsPtrPtr ->+convertEntry ::+  Ptr HerculesLoggerEntry ->+  -- | 'LogEntry' without timestamp+  IO LogEntry+convertEntry logEntryPtr = alloca \textStrPtr -> alloca \levelPtr -> alloca \activityIdPtr -> alloca \typePtr -> alloca \parentPtr -> alloca \fieldsPtrPtr ->   do     r <-       [C.throwBlock| int {         const HerculesLogger::LogEntry &ln = *$(HerculesLoggerEntry *logEntryPtr);-        *$(uint64_t *millisPtr) = ln.ms;         switch (ln.entryType) {           case 1:             *$(const char **textStrPtr) = strdup(ln.text.c_str());@@ -153,8 +157,9 @@             return 0;         }       }|]-    ms_ <- peek millisPtr+    -- Real timestamps and index number are set by the log monitoring process instead     let i_ = 0+        ms_ = 0     case r of       1 -> do         textStr <- peek textStrPtr@@ -259,15 +264,6 @@           yield lns           popper --- TODO: Use 'nubProgress' instead?---- | Remove spammy progress results.-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 () nubProgress = nubSubset (toChunk >=> toProgressKey)   where@@ -276,24 +272,6 @@     toChunk (Chunk a) = Just a     toChunk Flush = Nothing -unbatch :: (Monad m, Foldable l) => ConduitT (l a) (Flush a) m ()-unbatch = awaitForever $ \l -> do-  for_ l $ \a -> yield $ Chunk a-  yield Flush--batch :: (Monad m) => ConduitT (Flush a) [a] m ()-batch = go []-  where-    go acc =-      await >>= \case-        Nothing -> do-          unless (null acc) (yield $ reverse acc)-        Just Flush -> do-          unless (null acc) (yield $ reverse acc)-          go []-        Just (Chunk c) -> do-          go (c : acc)- nubSubset :: (Eq k, Monad m) => (a -> Maybe k) -> ConduitT a a m () nubSubset toKey =   await >>= \case@@ -317,71 +295,63 @@           yield a         nubSubset1 toKey ak -tryReadLine :: (MonadUnliftIO m) => Handle -> m (Either () ByteString)-tryReadLine s = tryJust (guard . isEOFError) (liftIO (BSC.hGetLine s))+data Handles = Handles {commands :: Handle, events :: Handle} -tapper :: (KatipContext m, MonadUnliftIO m) => TapState -> m ()-tapper s = do-  tryReadLine (readableStderrEnd s) >>= \case-    Left _ -> pass-    Right "__%%hercules terminate log%%__" -> pass-    Right ln -> do-      liftIO-        [C.throwBlock| void {-          std::string s = $bs-cstr:ln;-          herculesLogger->log(nix::lvlInfo, s);-        }|]-      tapper s+-- | Reassign file descriptors according to the scheme for the worker process.+--+-- __Before__+--+-- These are the roles assigned by the parent process:+--+--     * stdin: commands+--+--     * stdout: events+--+--     * stderr: arbitrary logs+--+-- __After__+--+--  Code in the child process can assume a pretty normal environment:+--+--     * stdin: empty+--+--     * stdout: arbitrary logs+--+--     * stderr: arbitrary logs+--+-- __Return__+--+-- The special handles that let us do structured communication with the parent process:+--+--    * commands+--+--    * events+--+-- __Output Buffering__+--+-- The standard handles are set to line buffering. The events handle is set to no buffering.+--+-- __Note__+--+-- This function is not interruptible and not thread-safe.+plumbWorkerStd :: IO Handles+plumbWorkerStd =+  Protolude.uninterruptibleMask \_unmask -> do+    -- in+    commandsHandle <- hDuplicate stdin+    devNull <- openFile "/dev/null" ReadMode+    hDuplicateTo devNull stdin --- | Like 'withTappableStderr' but takes care of the forking and waiting of--- the tapper thread.-withTappedStderr :: (KatipContext m, MonadUnliftIO m) => (TapState -> m ()) -> m a -> m a-withTappedStderr tapperFunction m = do-  (r, tapperDone) <--    withTappableStderr-      ( \tapState -> do-          tapperDone <- UnliftIO.Async.async (tapperFunction tapState)-          r <- m-          -- EOF doesn't seem to reach tapperFunction, so we simulate it with a magic line-          -- The preceding newline clears any unfinished lines but also introduces a needless newline in the final log-          hPutStrLn stderr ("\n__%%hercules terminate log%%__" :: ByteString)-          pure (r, tapperDone)-      )-  liftIO (timeout 30_000_000 $ wait tapperDone) >>= \case-    Just x -> pure x-    Nothing -> logLocM ErrorS "stderr thread did not finish in time"-  pure r+    -- out+    eventsHandle <- hDuplicate stdout+    hDuplicateTo stderr stdout -data TapState = TapState {originalStderrCopy :: Fd, readableStderrEnd :: Handle}+    -- err: unchanged --- | WARNING: Invoke only once. It leaks probably leaks a file descriptor or two.------ @withTappedStderr tapper mainAction@------ 'readableStderrEnd' receives EOF when your action terminates (unless you 'dup' the write end, aka 'stdError')-withTappableStderr :: (MonadUnliftIO m) => (TapState -> m a) -> m a-withTappableStderr = bracket (liftIO tapStderrPipe) (liftIO . revertTap)+    -- buffering+    hSetBuffering stdout LineBuffering+    hSetBuffering stderr LineBuffering -revertTap :: TapState -> IO ()-revertTap s = do-  void $ dupTo (originalStderrCopy s) stdError-  closeFd (originalStderrCopy s)+    hSetBuffering eventsHandle NoBuffering -tapStderrPipe :: IO TapState-tapStderrPipe = do-  oldStdError <- dup stdError-  (readable, writable) <- createPipe-  -- fd 2 := writable pipe-  void $ dupTo writable stdError-  -- Make sure all fds are set up correctly for the RTS's IO manager-  [oldStdError, readable, writable] & traverse_ \fd -> setNonBlockingFD (fromIntegral fd) True-  let mkHandle fd = do-        h <- fdToHandle fd-        hSetBuffering h LineBuffering-        pure h-  readableHandle <- mkHandle readable-  pure-    TapState-      { originalStderrCopy = oldStdError,-        readableStderrEnd = readableHandle-      }+    pure Handles {commands = commandsHandle, events = eventsHandle}

@@ -43,7 +43,12 @@  C.include "<nix/globals.hh>" +#if NIX_IS_AT_LEAST(2,19,0)+C.include "<nix/signals.hh>"+#else+-- redundant? C.include "<nix/fs-accessor.hh>"+#endif  #if NIX_IS_AT_LEAST(2,7,0) C.include "<nix/build-result.hh>"@@ -150,6 +155,7 @@       StorePath derivationPath = *$fptr-ptr:(nix::StorePath *derivationPath);        if ($(bool materializeDerivation)) {+        store.addTempRoot(derivationPath);         store.ensurePath(derivationPath);         auto derivation = store.derivationFromPath(derivationPath);         StorePathWithOutputs storePathWithOutputs { .path = derivationPath, .outputs = derivation.outputNames() };

@@ -1,78 +0,0 @@-{-# LANGUAGE BlockArguments #-}--module Hercules.Agent.Worker.Conduit where--import Data.Conduit (ConduitT, await, awaitForever, yield, (.|))-import Data.IORef (IORef, modifyIORef, newIORef, readIORef)-import Data.Sequence qualified 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

@@ -27,6 +27,7 @@         runEffectToken = Just $ Command.Effect.token command,         runEffectSecretsConfigPath = Just $ Command.Effect.secretsPath command,         runEffectServerSecrets = Command.Effect.serverSecrets command <&> fromViaJSON,+        runEffectConfiguredMountables = Command.Effect.configuredMountables command <&> fromViaJSON,         runEffectApiBaseURL = Command.Effect.apiBaseURL command,         runEffectDir = dir,         runEffectProjectId = Just $ Command.Effect.projectId command,@@ -42,12 +43,15 @@   let extraPaths = Command.Effect.inputDerivationOutputPaths command       drvPath = encodeUtf8 $ Command.Effect.drvPath command       ensureDeps = for_ extraPaths $ \input -> liftIO $ do-        CNix.ensurePath store =<< CNix.parseStorePath store input+        p <- CNix.parseStorePath store input+        CNix.addTemporaryRoot store p+        CNix.ensurePath store p   drvStorePath <- liftIO $ CNix.parseStorePath store drvPath   liftIO $ do     ensureDeps `catch` \e -> do       CNix.logInfo $ "while retrieving dependencies: " <> toS (displayException (e :: SomeException))       CNix.logInfo "unable to retrieve dependency; attempting fallback to local build"+      -- TODO read directly from cache?       CNix.ensurePath store drvStorePath       derivation <- CNix.getDerivation store drvStorePath       depDrvPaths <- CNix.getDerivationInputs store derivation@@ -60,6 +64,7 @@       Just drv -> pure drv       Nothing -> panic $ "Could not retrieve derivation " <> show drvPath <> " from local store or binary caches."   sources <- liftIO $ CNix.getDerivationSources store derivation-  for_ sources \src -> do-    liftIO $ CNix.ensurePath store src+  for_ sources \src -> liftIO do+    CNix.addTemporaryRoot store src+    CNix.ensurePath store src   pure derivation

@@ -3,11 +3,13 @@ import Control.Concurrent.STM (TVar) import Data.IORef (IORef) import Data.UUID (UUID)+import Data.Vector (Vector) import Hercules.Agent.Worker.HerculesStore (HerculesStore) import Hercules.Agent.WorkerProtocol.Command.BuildResult qualified as BuildResult import Hercules.Agent.WorkerProtocol.Event (Event) import Hercules.CNix (Ref) import Hercules.CNix.Store (Store, StorePath)+import Hercules.CNix.Verbosity qualified as CNix import Protolude  data HerculesState = HerculesState@@ -18,6 +20,7 @@     drvsInProgress :: IORef (Set StorePath),     herculesStore :: Ptr (Ref HerculesStore),     wrappedStore :: Store,-    shortcutChannel :: Chan (Maybe Event),-    extraNixOptions :: [(Text, Text)]+    sendEvents :: Vector Event -> IO (),+    extraNixOptions :: [(Text, Text)],+    nixVerbosity :: CNix.Verbosity   }

@@ -21,6 +21,7 @@ import Data.Map qualified as M import Data.Set qualified as S import Data.Text qualified as T+import Data.Type.Equality (type (~)) import Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent (OnPushHandlerEvent (OnPushHandlerEvent)) import Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent qualified import Hercules.API.Agent.Evaluate.EvaluateEvent.OnScheduleHandlerEvent (OnScheduleHandlerEvent (OnScheduleHandlerEvent), TimeConstraints (TimeConstraints))@@ -51,7 +52,7 @@ import Hercules.Agent.WorkerProtocol.Event.AttributeIFD qualified as Event.AttributeIFD import Hercules.Agent.WorkerProtocol.ViaJSON qualified as ViaJSON 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, initThread, isDerivation, isFunctor, match, rawValueType, rtValue, toRawValue, toValue, withEvalStateConduit)+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 Hercules.CNix.Expr.Raw qualified import Hercules.CNix.Expr.Schema (MonadEval, PSObject, dictionaryToMap, fromPSObject, provenance, requireDict, traverseArray, (#.), (#?), (#?!), (#??), ($?), (|!), type (->?), type (.))@@ -169,10 +170,11 @@   HerculesState ->   Eval ->   ConduitM i Event m ()-runEval st@HerculesState {herculesStore = hStore, shortcutChannel = shortcutChan, drvsCompleted = drvsCompl} eval = do+runEval st@HerculesState {herculesStore = hStore, drvsCompleted = drvsCompl} eval = do+  -- TODO: redundant as these have already been set?   for_ (Eval.extraNixOptions eval) $ liftIO . uncurry setGlobalOption   for_ (Eval.extraNixOptions eval) $ liftIO . uncurry setOption-  liftIO initThread+   let store = nixStore hStore       isFlake = Eval.isFlakeJob eval   s <- storeUri store@@ -218,12 +220,12 @@                 then do                   logLocM DebugS "Output already valid"                   -- Report IFD-                  liftIO $ writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) Nothing False+                  liftIO $ st.sendEvents $ pure $ 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+                  liftIO $ st.sendEvents $ pure $ Event.Build drvPath (decode outputName) Nothing doBlock                    buildAsync <- liftIO $ asyncInTVarMap (drvStorePath, outputName) (drvOutputSubstituteAsyncs st) do                     ensurePath (wrappedStore st) outputPath@@ -239,7 +241,7 @@                         clearSubstituterCaches                         clearPathInfoCache store                         ensurePath (wrappedStore st) outputPath `catch` \(_e1 :: SomeException) -> do-                          writeChan shortcutChan $ Just $ Event.Build drvPath (decode outputName) (Just attempt0) doBlock+                          st.sendEvents $ pure $ Event.Build drvPath (decode outputName) (Just attempt0) doBlock                            (_, result') <-                             wait =<< asyncInTVarMap drvStorePath (drvRebuildAsyncs st) do@@ -599,7 +601,8 @@                 then walkDerivation store evalState False path attrValue                 else do                   attrs <- liftIO $ getAttrs evalState attrValue-                  void $+                  recurse <- shouldWalkIntoAttrsAndLog path evalState attrs+                  when recurse . void $                     flip M.traverseWithKey attrs $                       \name value ->                         if depthRemaining > 0@@ -611,7 +614,7 @@                                   treeWorkAttrPath = path ++ [name],                                   treeWorkThunk = value                                 }-                          else yield (Event.Error $ "Max recursion depth reached at path " <> show path)+                          else yield (Event.Error $ "Max recursion depth reached at path " <> show (path ++ [name]))             _any -> do               vt <- liftIO $ rawValueType v               unless@@ -629,6 +632,23 @@           treeWorkRemainingDepth = 10,           treeWorkThunk = initialThunk         }++shouldWalkIntoAttrsAndLog :: (KatipContext m) => [ByteString] -> Ptr EvalState -> Map ByteString RawValue -> m Bool+shouldWalkIntoAttrsAndLog path evalState attrs =+  case M.lookup "_type" attrs of+    Just v -> do+      -- `_type` value should be cheap, so let's log it+      liftIO (match evalState v) >>= \case+        Right (IsString sv) -> do+          s <- liftIO $ getStringIgnoreContext sv+          logLocM DebugS $ logStr ("Ignoring " <> show path <> " : _type = " <> show s :: Text)+        Right _ -> do+          logLocM DebugS $ logStr ("Ignoring " <> show path <> " : _type is not a string" :: Text)+        Left _ -> do+          logLocM DebugS $ logStr ("Ignoring " <> show path <> " : _type contains error" :: Text)+      pure False+    Nothing -> do+      pure True  withIFDQueue ::   (MonadUnliftIO m) =>

@@ -7,14 +7,14 @@ import UnliftIO (MonadUnliftIO) import UnliftIO.Exception (bracket) -withKatip :: (MonadUnliftIO m) => KatipContextT m a -> m a-withKatip m = do+withKatip :: (MonadUnliftIO m) => Severity -> KatipContextT m a -> m a+withKatip s m = do   let format :: forall a. (LogItem a) => ItemFormatter a       format = (\_ _ _ -> "@katip ") <> jsonFormat   -- Use a duplicate of stderr, to make sure we keep logging there, even after   -- we reassign stderr to catch output from git and other subprocesses of Nix.   dupStderr <- liftIO (fdToHandle =<< dup stdError)-  handleScribe <- liftIO $ mkHandleScribeWithFormatter format (ColorLog False) dupStderr (permitItem DebugS) V2+  handleScribe <- liftIO $ mkHandleScribeWithFormatter format (ColorLog False) dupStderr (permitItem s) V2   let makeLogEnv = registerScribe "stderr" handleScribe defaultScribeSettings =<< initLogEnv "Worker" "production"       initialContext = ()       extraNs = mempty -- "Worker" is already set in initLogEnv.

@@ -1,7 +1,7 @@ cabal-version: 2.4  name:           hercules-ci-agent-version:        0.9.13+version:        0.10.1 synopsis:       Runs Continuous Integration tasks on your machines category:       Nix, CI, Testing, DevOps homepage:       https://docs.hercules-ci.com@@ -35,24 +35,24 @@  common hs   default-extensions: -    NoImplicitPrelude     DeriveGeneric     DeriveTraversable     DerivingStrategies     DisambiguateRecordFields     FlexibleContexts-    ImportQualifiedPost     InstanceSigs     LambdaCase     MultiParamTypeClasses-    NumericUnderscores-    OverloadedRecordDot+    NoImplicitPrelude     OverloadedStrings     RankNTypes     TupleSections     TypeApplications     TypeOperators-+    OverloadedRecordDot+    ImportQualifiedPost+    NumericUnderscores+    ScopedTypeVariables   ghc-options: -Wall -fwarn-tabs -fwarn-unused-imports -fwarn-missing-signatures -fwarn-name-shadowing -fwarn-incomplete-patterns -Werror=incomplete-patterns -Werror=missing-fields  common rts@@ -105,6 +105,7 @@       Data.Fixed.Extras       Data.Time.Extras       Hercules.Agent.Binary+      Hercules.Agent.Conduit       Hercules.Agent.NixFile       Hercules.Agent.NixFile.GitSource       Hercules.Agent.NixFile.HerculesCIArgs@@ -114,6 +115,7 @@       Hercules.Agent.Sensitive       Hercules.Agent.Socket       Hercules.Agent.STM+      Hercules.Agent.Store       Hercules.Agent.WorkerProcess       Hercules.Agent.WorkerProtocol.ViaJSON       Hercules.Agent.WorkerProtocol.Command@@ -126,8 +128,9 @@       Hercules.Agent.WorkerProtocol.Event.AttributeError       Hercules.Agent.WorkerProtocol.Event.AttributeIFD       Hercules.Agent.WorkerProtocol.Event.BuildResult-      Hercules.Agent.WorkerProtocol.LogSettings       Hercules.Agent.WorkerProtocol.Orphans+      Hercules.Agent.WorkerProtocol.OutputInfo+      Hercules.Agent.WorkerProtocol.WorkerConfig       Hercules.Effect       Hercules.Effect.Container       Hercules.Secrets@@ -140,8 +143,6 @@       Paths_hercules_ci_agent   hs-source-dirs:       src-  default-extensions: DeriveGeneric DeriveTraversable DisambiguateRecordFields FlexibleContexts InstanceSigs LambdaCase MultiParamTypeClasses NoImplicitPrelude OverloadedStrings RankNTypes TupleSections TypeApplications TypeOperators-  ghc-options: -Wall -fwarn-tabs -fwarn-unused-imports -fwarn-missing-signatures -fwarn-name-shadowing -fwarn-incomplete-patterns   build-depends:       aeson >= 2     , async@@ -176,6 +177,7 @@     , tagged     , temporary     , text+    , tls     , time     , transformers     , transformers-base@@ -184,6 +186,7 @@     , unliftio-core     , unliftio     , uuid+    , vector     , websockets     , wuss   default-language: Haskell2010@@ -207,6 +210,9 @@       Hercules.Agent.Client       Hercules.Agent.Config       Hercules.Agent.Config.BinaryCaches+      Hercules.Agent.Config.Combined+      Hercules.Agent.Config.Json+      Hercules.Agent.Config.Toml       Hercules.Agent.Compat       Hercules.Agent.Effect       Hercules.Agent.Env@@ -215,7 +221,10 @@       Hercules.Agent.Evaluate.TraversalQueue       Hercules.Agent.Files       Hercules.Agent.Init+      Hercules.Agent.InitWorkerConfig       Hercules.Agent.Log+      Hercules.Agent.LogSocket+      Hercules.Agent.Memo       Hercules.Agent.Netrc       Hercules.Agent.Netrc.Env       Hercules.Agent.Nix@@ -223,6 +232,7 @@       Hercules.Agent.Nix.Init       Hercules.Agent.Nix.RetrieveDerivationInfo       Hercules.Agent.Options+      Hercules.Agent.ResourceLimiter       Hercules.Agent.SecureDirectory       Hercules.Agent.ServiceInfo       Hercules.Agent.Token@@ -231,14 +241,13 @@       Paths_hercules_ci_agent   hs-source-dirs:       hercules-ci-agent-  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=-maxN8 -qg"   build-depends:       aeson     , async     , attoparsec     , base     , base64-bytestring+    , bifunctors     , binary     , binary-conduit     , bytestring@@ -258,8 +267,8 @@     , filepath     , hercules-ci-agent     , hercules-ci-api-    , hercules-ci-api-core == 0.1.5.1-    , hercules-ci-api-agent == 0.5.0.1+    , hercules-ci-api-core == 0.1.6.0+    , hercules-ci-api-agent == 0.5.1.0     , hostname     , http-client     , http-client-tls@@ -276,6 +285,7 @@     , optparse-applicative     , process     , process-extras+    , profunctors     , protolude     , safe-exceptions     , scientific@@ -309,7 +319,6 @@       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@@ -326,15 +335,7 @@       cbits/hercules-logger.cxx       cbits/nix-2.4/hercules-store.cxx -  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=-maxN8@@ -399,15 +400,23 @@   main-is: TestMain.hs   other-modules:       Data.Conduit.Extras+      Hercules.Agent.Conduit       Hercules.Agent.Log+      Hercules.Agent.Config+      Hercules.Agent.Config.Combined+      Hercules.Agent.Config.Json+      Hercules.Agent.Config.Toml+      Hercules.Agent.ConfigSpec       Hercules.Agent.NixPath       Hercules.Agent.NixPathSpec       Hercules.Agent.Nix.RetrieveDerivationInfo       Hercules.Agent.Nix.RetrieveDerivationInfoSpec       Hercules.Agent.WorkerProcess       Hercules.Agent.WorkerProcessSpec-      Hercules.Agent.Worker.Conduit-      Hercules.Agent.Worker.ConduitSpec+      Hercules.Agent.WorkerProtocol.ViaJSON+      Hercules.Agent.WorkerProtocol.WorkerConfig+      Hercules.Agent.WorkerProtocol.WorkerConfigSpec+      Hercules.Agent.ConduitSpec       Hercules.Agent.Worker.STM       Hercules.Agent.Worker.STMSpec       Hercules.Secrets@@ -421,13 +430,12 @@       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:       aeson     , async     , attoparsec     , base+    , bifunctors     , binary     , binary-conduit     , bytestring@@ -440,20 +448,30 @@     , hercules-ci-agent     , hercules-ci-api-core     , hspec+    , HUnit     , katip+    , lens+    , lens-aeson     , lifted-async     , lifted-base     , monad-control     , mtl     , process+    , profunctors     , protolude+    , QuickCheck     , safe-exceptions+    , scientific     , stm     , tagged     , temporary     , text+    , tomland+    , transformers     , transformers-base     , unliftio-core+    , unordered-containers+    , uuid     , vector   build-tool-depends:       hspec-discover:hspec-discover

@@ -38,9 +38,6 @@ import Hercules.Agent.Build qualified as Build import Hercules.Agent.CabalInfo (herculesAgentVersion) import Hercules.Agent.Cache qualified as Cache-#if ! MIN_VERSION_cachix(1, 4, 0) || MIN_VERSION_cachix(1, 5, 0)-import Hercules.Agent.Cachix.Env qualified as Cachix.Env-#endif import Hercules.Agent.Client   ( lifeCycleClient,     tasksClient,@@ -56,10 +53,13 @@ import Hercules.Agent.Evaluate qualified as Evaluate import Hercules.Agent.Init qualified as Init import Hercules.Agent.Log+import Hercules.Agent.LogSocket (LogSettings (LogSettings), withLoggerNoFlush)+import Hercules.Agent.LogSocket qualified import Hercules.Agent.Netrc qualified as Netrc import Hercules.Agent.Options qualified as Options import Hercules.Agent.STM-import Hercules.Agent.Socket as Socket+import Hercules.Agent.ServiceInfo qualified+import Hercules.Agent.Socket qualified as Socket import Hercules.Agent.Token (withAgentToken) import Hercules.CNix.Store qualified as CNix.Store import Hercules.Error@@ -67,6 +67,7 @@     exponential,     retry,   )+import Network.URI (uriToString) import Protolude hiding   ( atomically,     bracket,@@ -117,23 +118,40 @@               withAgentSocket hello tasks \socket ->                 withApplicationLevelPinger socket $ do                   logLocM InfoS "Agent online."+                  storeProtocolVersion <- liftIO do+                    CNix.Store.withStore CNix.Store.getStoreProtocolVersion++                  let baseURL = env.serviceInfo.bulkSocketBaseURL & toString+                      toString uri = uriToString identity uri "" & toS++                  let launchTask' task path_ f = launchTask tasks socket (Task.upcastId task.id) do+                        Netrc.withNixNetrc $ Cache.withCaches do+                          withLoggerNoFlush+                            ("task" <> show task.id)+                            storeProtocolVersion+                            LogSettings+                              { path = path_,+                                baseURL = baseURL,+                                token = task.logToken+                              }+                            \logger -> do+                              f logger+                   forever $ do-                    liftIO (atomically $ readTChan $ serviceChan socket) >>= \case+                    liftIO (atomically $ readTChan $ socket.serviceChan) >>= \case                       ServicePayload.ServiceInfo _ -> pass                       ServicePayload.StartEvaluation evalTask ->-                        launchTask tasks socket (Task.upcastId $ EvaluateTask.id evalTask) do+                        launchTask' evalTask "/api/v1/logs/build/socket" \logger -> do                           Netrc.withNixNetrc $ Cache.withCaches do                             CNix.Store.withStore \store -> do-                              Evaluate.performEvaluation store evalTask+                              Evaluate.performEvaluation logger store evalTask                             pure $ TaskStatus.Successful ()                       ServicePayload.StartBuild buildTask ->-                        launchTask tasks socket (Task.upcastId $ BuildTask.id buildTask) do-                          Netrc.withNixNetrc $ Cache.withCaches do-                            Build.performBuild buildTask+                        launchTask' buildTask "/api/v1/logs/build/socket" \logger -> do+                          Build.performBuild logger buildTask                       ServicePayload.StartEffect effectTask ->-                        launchTask tasks socket (Task.upcastId $ EffectTask.id effectTask) do-                          Netrc.withNixNetrc $ Cache.withCaches do-                            Effect.performEffect effectTask+                        launchTask' effectTask "/api/v1/logs/build/socket" \logger -> do+                          Effect.performEffect logger effectTask                       ServicePayload.Cancel cancellation -> cancelTask tasks socket cancellation  withTaskState :: (TVar (Map (Id (Task Task.Any)) ThreadId) -> App a) -> App a@@ -151,7 +169,7 @@         now <- readTVar tasks         guard $ now /= current         pure now-      katipAddContext (sl "state" (A.toJSON $ fmap Prelude.show new)) do+      katipAddContext (sl "state" (A.toJSON $ fmap Prelude.show new) <> sl "currentConcurrency" (length new)) do         logLocM DebugS "Task state updated"       watchChange new 

@@ -1,20 +1,24 @@+{-# LANGUAGE BlockArguments #-} {-# LANGUAGE CPP #-}+ module Hercules.Agent.Build where  import Data.Aeson qualified as A import Data.IORef.Lifted import Data.Map qualified as M+import Data.Vector (Vector) import Hercules.API.Agent.Build qualified as API.Build import Hercules.API.Agent.Build.BuildEvent qualified as BuildEvent-import Hercules.API.Agent.Build.BuildEvent.OutputInfo+import Hercules.API.Agent.OutputInfo   ( OutputInfo,   )-import Hercules.API.Agent.Build.BuildEvent.OutputInfo qualified as OutputInfo+import Hercules.API.Agent.OutputInfo qualified as OutputInfo import Hercules.API.Agent.Build.BuildEvent.Pushed qualified as Pushed import Hercules.API.Agent.Build.BuildTask   ( BuildTask,   ) import Hercules.API.Agent.Build.BuildTask qualified as BuildTask+import Hercules.API.Logs.LogEntry (LogEntry) import Hercules.API.Servant (noContent) import Hercules.API.TaskStatus (TaskStatus) import Hercules.API.TaskStatus qualified as TaskStatus@@ -26,29 +30,27 @@ import Hercules.Agent.Config qualified as Config import Hercules.Agent.Env import Hercules.Agent.Env qualified as Env+import Hercules.Agent.InitWorkerConfig qualified as InitWorkerConfig import Hercules.Agent.Log-import Hercules.Agent.Nix qualified as Nix-import Hercules.Agent.Sensitive (Sensitive (Sensitive))-import Hercules.Agent.ServiceInfo qualified as ServiceInfo import Hercules.Agent.WorkerProcess import Hercules.Agent.WorkerProcess qualified as WorkerProcess import Hercules.Agent.WorkerProtocol.Command qualified as Command import Hercules.Agent.WorkerProtocol.Command.Build qualified as Command.Build import Hercules.Agent.WorkerProtocol.Event qualified as Event import Hercules.Agent.WorkerProtocol.Event.BuildResult qualified as BuildResult-import Hercules.Agent.WorkerProtocol.LogSettings qualified as LogSettings+import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON (ViaJSON)) import Hercules.CNix.Store qualified as CNix import Hercules.Error (defaultRetry)-import Network.URI qualified import Protolude import System.Process+import qualified Hercules.Agent.WorkerProtocol.OutputInfo as Proto -performBuild :: BuildTask.BuildTask -> App TaskStatus-performBuild buildTask = do+performBuild :: (Vector LogEntry -> IO ()) -> BuildTask.BuildTask -> App TaskStatus+performBuild sendLogEntries buildTask = katipAddContext (sl "taskDerivationPath" buildTask.derivationPath) $ do   workerExe <- getWorkerExe   commandChan <- liftIO newChan   statusRef <- newIORef Nothing-  extraNixOptions <- Nix.askExtraOptions+  workerConfig <- InitWorkerConfig.getWorkerConfig   workerEnv <-     liftIO $       WorkerProcess.prepareEnv@@ -57,22 +59,24 @@               extraEnv = mempty             }         )-  let opts = [show extraNixOptions]-      procSpec =-        (System.Process.proc workerExe opts)+  let procSpec =+        (System.Process.proc workerExe ["build", toS buildTask.derivationPath])           { env = Just workerEnv,             close_fds = True,             cwd = Nothing           }       writeEvent :: Event.Event -> App ()       writeEvent event = case event of+        Event.LogItems (ViaJSON e) -> do+          liftIO (sendLogEntries e)         Event.BuildResult r -> writeIORef statusRef $ Just r         Event.Exception e -> do           logLocM DebugS $ logStr (show e :: Text)           panic e         _ -> pass-  baseURL <- asks (ServiceInfo.bulkSocketBaseURL . Env.serviceInfo)   materialize <- asks (not . Config.nixUserIsTrusted . Env.config)+  -- Remote builds need a whole drv closure.+  materializePlatforms <- asks (Config.remotePlatformsWithSameFeatures . Env.config)   liftIO $     writeChan commandChan $       Just $@@ -80,23 +84,19 @@           Command.Build.Build             { drvPath = BuildTask.derivationPath buildTask,               inputDerivationOutputPaths = encodeUtf8 <$> BuildTask.inputDerivationOutputPaths buildTask,-              logSettings =-                LogSettings.LogSettings-                  { token = Sensitive $ BuildTask.logToken buildTask,-                    path = "/api/v1/logs/build/socket",-                    baseURL = toS $ Network.URI.uriToString identity baseURL ""-                  },-              materializeDerivation = materialize+              materializeDerivation = materialize,+              materializePlatforms = materializePlatforms & fromMaybe [] <&> encodeUtf8             }   let stderrHandler =         stderrLineHandler+          sendLogEntries           ( M.fromList               [ ("taskId", A.toJSON (BuildTask.id buildTask)),                 ("derivationPath", A.toJSON (BuildTask.derivationPath buildTask))               ]           )           "Builder"-  exitCode <- runWorker procSpec stderrHandler commandChan writeEvent+  exitCode <- runWorker workerConfig procSpec (stderrHandler) commandChan writeEvent   logLocM DebugS $ "Worker exit: " <> logStr (show exitCode :: Text)   case exitCode of     ExitSuccess -> pass@@ -113,18 +113,26 @@ #endif       reportSuccess buildTask       pure $ TaskStatus.Successful ()-    Just BuildResult.BuildFailure {} -> pure $ TaskStatus.Terminated ()+    Just BuildResult.BuildFailure {errorMessage = errorMessage} ->+      katipAddContext (sl "errorMessage" errorMessage) do+        logLocM DebugS "Build failed"+        pure $ TaskStatus.Terminated ()     Nothing -> pure $ TaskStatus.Exceptional "Build did not complete" -convertOutputs :: Text -> [BuildResult.OutputInfo] -> Map Text OutputInfo+convertOutputs :: Text -> [Proto.OutputInfo] -> Map Text OutputInfo convertOutputs deriver = foldMap $ \oi ->-  M.singleton (decodeUtf8With lenientDecode $ BuildResult.name oi) $-    OutputInfo.OutputInfo+  M.singleton (decodeUtf8With lenientDecode oi.name) $+    convertOutputInfo deriver oi++convertOutputInfo :: Text -> Proto.OutputInfo -> OutputInfo+convertOutputInfo deriver oi =+  OutputInfo.OutputInfo       { OutputInfo.deriver = deriver,-        name = decodeUtf8With lenientDecode $ BuildResult.name oi,-        path = decodeUtf8With lenientDecode $ BuildResult.path oi,-        size = fromIntegral $ BuildResult.size oi,-        hash = decodeUtf8With lenientDecode $ BuildResult.hash oi+        name = decodeUtf8With lenientDecode oi.name,+        path = decodeUtf8With lenientDecode oi.path,+        size = fromIntegral oi.size,+        hash = decodeUtf8With lenientDecode oi.hash,+        references = Just (decodeUtf8With lenientDecode <$> oi.references)       }  push :: CNix.Store -> BuildTask -> Map Text OutputInfo -> App ()

@@ -38,6 +38,13 @@     ]     m +getConfiguredSubstituters :: App [Text]+getConfiguredSubstituters = do+  nixCaches <- asks (Config.nixCaches . Env.binaryCaches)+  let substs = nixCaches & toList <&> NixCache.storeURI+  csubsts <- Cachix.getSubstituters+  pure (substs <> csubsts)+ push ::   -- | Nix Store   CNix.Store ->

@@ -84,6 +84,7 @@   paths' <- paths & traverse (convertPath nixStore) #else   let paths' = paths+      _ = nixStore -- silence unused warning #endif   void $     Cachix.Push.pushClosure

@@ -24,7 +24,8 @@ import Cachix.Client.URI (defaultCachixBaseUrl) import Control.Monad.IO.Unlift (MonadUnliftIO) import Data.Map qualified as M-import Hercules.Agent.Cachix.Env as Env+import Hercules.Agent.Cachix.Env (PushCache (PushCache))+import Hercules.Agent.Cachix.Env qualified as Env import Hercules.Agent.Config qualified as Config import Hercules.Error import Hercules.Formats.CachixCache qualified as CachixCache

@@ -31,18 +31,6 @@ import Servant.Client.Streaming (ClientM) import Servant.Client.Streaming qualified --- | Bad instance to make it the client for State api compile. GHC seems to pick--- the wrong overlappable instance.-instance-  FromSourceIO-    ByteString-    ( Headers-        '[Hercules.API.Agent.State.ContentLength]-        (SourceIO ByteString)-    )-  where-  fromSourceIO = addHeader (-1) . fromSourceIO- client :: AgentAPI ClientAuth (AsClientT ClientM) client = fromServant $ Servant.Client.Streaming.client (servantApi @ClientAuth) 

@@ -1,35 +1,46 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralisedNewtypeDeriving #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wno-orphans #-}  module Hercules.Agent.Config   ( Config (..),     FinalConfig,     ConfigPath (..),     Purpose (..),+    Mountable (..),     readConfig,     finalizeConfig,++    -- * Export for testing+    combiCodec,   ) where  import Data.Aeson qualified as A-import Data.Aeson.KeyMap qualified as AK-import Data.Scientific (floatingOrInteger, fromFloatDigits)-import Data.Vector qualified as V+import Data.Aeson.Types qualified as A+import Data.Profunctor (Star (Star)) import GHC.Conc (getNumProcessors)+import Hercules.Agent.Config.Combined+import Hercules.Agent.Config.Json as Json+import Hercules.Agent.Config.Toml qualified as Toml import Hercules.CNix.Verbosity (Verbosity (..))+import Hercules.Formats.Mountable (Mountable (Mountable))+import Hercules.Formats.Mountable qualified as Mountable import Katip (Severity (..)) import Protolude hiding (to) import System.Environment qualified import System.FilePath ((</>))-import Toml+import Toml (Key)+import Toml qualified -data ConfigPath = TomlPath FilePath+newtype ConfigPath = ConfigPath FilePath  nounPhrase :: ConfigPath -> Text-nounPhrase (TomlPath p) = "your agent.toml file from " <> show p+nounPhrase (ConfigPath p) = "your agent config file from " <> show p  data Purpose = Input | Final @@ -59,87 +70,86 @@     logLevel :: Item purpose 'Required Severity,     nixVerbosity :: Item purpose 'Required Verbosity,     labels :: Item purpose 'Required (Map Text A.Value),-    allowInsecureBuiltinFetchers :: Item purpose 'Required Bool+    allowInsecureBuiltinFetchers :: Item purpose 'Required Bool,+    remotePlatformsWithSameFeatures :: Item purpose 'Optional [Text],+    effectMountables :: Map Text Mountable   }   deriving (Generic)  deriving instance Show (Config 'Final) -tomlCodec :: TomlCodec (Config 'Input)-tomlCodec =+combiCodec :: Combi' (Config 'Input)+combiCodec =   Config-    <$> dioptional (Toml.text "apiBaseUrl")-      .= herculesApiBaseURL-    <*> dioptional (Toml.bool "nixUserIsTrusted")-      .= nixUserIsTrusted-    <*> dioptional-      ( Toml.dimatch matchRight Right (Toml.int "concurrentTasks")-          <|> Toml.dimatch matchLeft Left (Toml.textBy (\() -> "auto") isAuto "concurrentTasks")+    <$> opt (textAtKey "apiBaseUrl") .=. herculesApiBaseURL+    <*> opt (boolAtKey "nixUserIsTrusted") .=. nixUserIsTrusted+    <*> opt+      ( combi+          ( Toml.dimatch matchRight Right (Toml.int "concurrentTasks")+              <|> Toml.dimatch matchLeft Left (Toml.textBy (\() -> "auto") isAuto "concurrentTasks")+          )+          ( Json.dimatch matchRight Right (Json.int "concurrentTasks")+              <|> Json.dimatch matchLeft Left (Json.textBy (\() -> "auto") isAuto "concurrentTasks")+          )       )-      .= concurrentTasks-    <*> dioptional (Toml.string keyBaseDirectory)-      .= baseDirectory-    <*> dioptional (Toml.string "staticSecretsDirectory")-      .= staticSecretsDirectory-    <*> dioptional (Toml.string "workDirectory")-      .= workDirectory-    <*> dioptional (Toml.string keyClusterJoinTokenPath)-      .= clusterJoinTokenPath-    <*> dioptional (Toml.string "binaryCachesPath")-      .= binaryCachesPath-    <*> dioptional (Toml.string "secretsJsonPath")-      .= secretsJsonPath-    <*> dioptional (Toml.enumBounded "logLevel")-      .= logLevel-    <*> dioptional (Toml.enumBounded "nixVerbosity")-      .= nixVerbosity-    <*> dioptional (Toml.tableMap _KeyText embedJson "labels")-      .= labels-    <*> dioptional (Toml.bool "allowInsecureBuiltinFetchers")-      .= allowInsecureBuiltinFetchers--embedJson :: Key -> TomlCodec A.Value-embedJson key =-  Codec-    { codecRead =-        codecRead (match (embedJsonBiMap key) key)-          <!> codecRead (A.Object . AK.fromHashMapText <$> Toml.tableHashMap _KeyText embedJson key),-      codecWrite = panic "embedJson.write: not implemented" $ \case-        A.String s -> A.String <$> codecWrite (Toml.text key) s-        A.Number sci -> A.Number . fromRational . toRational <$> codecWrite (Toml.double key) (fromRational $ toRational sci)-        A.Bool b -> A.Bool <$> codecWrite (Toml.bool key) b-        A.Array a -> A.Array . V.fromList <$> codecWrite (Toml.arrayOf (embedJsonBiMap key) key) (Protolude.toList a)-        A.Object o -> A.Object . AK.fromHashMapText <$> codecWrite (Toml.tableHashMap _KeyText embedJson key) (AK.toHashMapText o)-        A.Null -> eitherToTomlState (Left ("null is not supported in TOML" :: Text))-    }--embedJsonBiMap :: Key -> TomlBiMap A.Value AnyValue-embedJsonBiMap _key =-  -- TODO: use key for error reporting-  BiMap-    { forward = panic "embedJsonBiMap.forward: not implemented" $ \case-        A.String s -> pure $ AnyValue $ Text s-        A.Number sci -> case floatingOrInteger sci of-          Left fl -> pure $ AnyValue $ Double fl -- lossy-          Right i -> pure $ AnyValue $ Integer i-        A.Bool b -> pure $ AnyValue $ Bool b-        A.Array _a -> Left $ ArbitraryError "Conversion from JSON array of arrays to TOML not implemented yet"-        A.Object _o -> Left $ ArbitraryError "Conversion from JSON array of objects to TOML is not supported"-        A.Null -> Left $ ArbitraryError "JSON null is not supported in TOML",-      backward = anyValueToJSON-    }+      .=. concurrentTasks+    <*> opt (stringAtKey keyBaseDirectory) .=. baseDirectory+    <*> opt (stringAtKey "staticSecretsDirectory") .=. staticSecretsDirectory+    <*> opt (stringAtKey "workDirectory") .=. workDirectory+    <*> opt (stringAtKey keyClusterJoinTokenPath) .=. clusterJoinTokenPath+    <*> opt (stringAtKey "binaryCachesPath") .=. binaryCachesPath+    <*> opt (stringAtKey "secretsJsonPath") .=. secretsJsonPath+    <*> opt (enumBoundedAtKey "logLevel") .=. logLevel+    <*> opt (enumBoundedAtKey "nixVerbosity") .=. nixVerbosity+    <*> opt+      ( combi+          (Toml.tableMap Toml._KeyText Toml.embedJson "labels")+          (Json.tableMap' Json.value' "labels")+      )+      .=. labels+    <*> opt (boolAtKey "allowInsecureBuiltinFetchers") .=. allowInsecureBuiltinFetchers+    <*> opt+      ( combi+          (Toml.arrayOf Toml._Text "remotePlatformsWithSameFeatures")+          (Json.arrayOf Json._Text "remotePlatformsWithSameFeatures")+      )+      .=. remotePlatformsWithSameFeatures+    <*> optEmpty+      ( combi+          (Toml.tableMap Toml._KeyText (Toml.table (forToml mountableCodec)) "effectMountables")+          (Json.tableMap' (forJson mountableCodec) "effectMountables")+      )+      .=. effectMountables -anyValueToJSON :: AnyValue -> Either TomlBiMapError A.Value-anyValueToJSON = \case-  AnyValue (Bool b) -> pure (A.Bool b)-  AnyValue (Integer i) -> pure (A.Number $ fromIntegral i)-  AnyValue (Double d) -> pure (A.Number $ fromFloatDigits d)-  AnyValue (Text t) -> pure (A.String t)-  AnyValue (Zoned _zt) -> Left (ArbitraryError "Conversion from TOML zoned time to JSON not implemented yet. Use a string.")-  AnyValue (Local _zt) -> Left (ArbitraryError "Conversion from TOML local time to JSON not implemented yet. Use a string.")-  AnyValue (Day _d) -> Left (ArbitraryError "Conversion from TOML day to JSON not implemented yet. Use a string.")-  AnyValue (Hours _h) -> Left (ArbitraryError "Conversion from TOML hours to JSON not implemented yet. Use a string.")-  AnyValue (Array a) -> A.Array <$> sequence (V.fromList (a <&> AnyValue <&> anyValueToJSON))+mountableCodec :: Combi' Mountable+mountableCodec =+  Mountable+    <$> textAtKey "source" .=. Mountable.source+    <*> boolAtKey "readOnly" .=. Mountable.readOnly+    <*> combi+      ((throwImmediately . A.parseJSON) <$> Toml.embedJson "condition")+      ( GCodec+          ( do+              let parse =+                    A.withObject "Mountable" $ \o ->+                      o A..: "condition"+              v <- ask+              -- seq (Debug.Trace.traceShowId $ Debug.Trace.trace "------------" v) pass+              case A.parseEither parse v of+                Right x -> pure x+                Left e -> throwError $ "Error parsing condition for mountable: " <> show e+          )+          (Star $ \_ -> panic "condition toJSON not implemented")+      )+      .=. (A.toJSON . Mountable.condition)+  where+    -- "Parser" is more like "parse result"+    -- TODO: more friendly reporting?+    throwImmediately :: A.Parser a -> a+    throwImmediately p =+      A.parseEither (const p) () & \case+        Left e -> panic $ "Error parsing condition for mountable: " <> show e+        Right a -> a  matchLeft :: Either a b -> Maybe a matchLeft (Left a) = Just a@@ -170,7 +180,8 @@  readConfig :: ConfigPath -> IO (Config 'Input) readConfig loc = case loc of-  TomlPath fp -> Toml.decodeFile tomlCodec (toS fp)+  ConfigPath fp | ".json" `isSuffixOf` fp -> Json.decodeFile (forJson combiCodec) (toS fp)+  ConfigPath fp -> Toml.decodeFile (forToml combiCodec) (toS fp)  finalizeConfig :: ConfigPath -> Config 'Input -> IO (Config 'Final) finalizeConfig loc input = do@@ -220,5 +231,7 @@         logLevel = logLevel input & fromMaybe InfoS,         nixVerbosity = nixVerbosity input & fromMaybe Talkative,         labels = fromMaybe mempty $ labels input,-        allowInsecureBuiltinFetchers = fromMaybe False $ allowInsecureBuiltinFetchers input+        allowInsecureBuiltinFetchers = fromMaybe False $ allowInsecureBuiltinFetchers input,+        remotePlatformsWithSameFeatures = remotePlatformsWithSameFeatures input,+        effectMountables = effectMountables input       }

@@ -0,0 +1,48 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Hercules.Agent.Config.Combined where++import Data.Bifunctor.Product (Product (Pair))+import Data.Profunctor (Profunctor (..))+import Hercules.Agent.Config.Json (JsonCodec)+import Hercules.Agent.Config.Json qualified as Json+import Hercules.Agent.Config.Toml ()+import Protolude+import Toml (Key)+import Toml qualified++newtype Combi a b = Combi (Data.Bifunctor.Product.Product (Toml.Codec) JsonCodec a b)+  deriving newtype (Functor, Profunctor)++type Combi' a = Combi a a++combi :: Toml.Codec a b -> JsonCodec a b -> Combi a b+combi toml json = Combi $ Pair toml json++forJson :: Combi a b -> JsonCodec a b+forJson (Combi (Data.Bifunctor.Product.Pair _ b)) = b++forToml :: Combi a b -> Toml.Codec a b+forToml (Combi (Data.Bifunctor.Product.Pair a _)) = a++instance Applicative (Combi a) where+  pure a = Combi (Data.Bifunctor.Product.Pair (pure a) (pure a))+  Combi (Data.Bifunctor.Product.Pair f g) <*> Combi (Data.Bifunctor.Product.Pair a b) = Combi (Data.Bifunctor.Product.Pair (f <*> a) (g <*> b))++textAtKey :: Key -> Combi' Text+textAtKey k = Combi (Data.Bifunctor.Product.Pair (Toml.text k) (Json.text k))++stringAtKey :: Key -> Combi' [Char]+stringAtKey k = Combi (Data.Bifunctor.Product.Pair (Toml.string k) (Json.string k))++boolAtKey :: Key -> Combi' Bool+boolAtKey k = Combi (Data.Bifunctor.Product.Pair (Toml.bool k) (Json.bool k))++opt :: Combi' a -> Combi' (Maybe a)+opt x = Combi $ Pair (Toml.dioptional (forToml x)) (Json.proOptional (forJson x))++optEmpty :: (Monoid a) => Combi' a -> Combi' a+optEmpty c = dimap Just (fromMaybe mempty) (opt c)++enumBoundedAtKey :: (Bounded a, Enum a, Show a) => Key -> Combi' a+enumBoundedAtKey k = Combi (Data.Bifunctor.Product.Pair (Toml.enumBounded k) (Json.enumBounded k))

@@ -0,0 +1,283 @@+{-# LANGUAGE ScopedTypeVariables #-}++-- | JSON codecs using the tomland method+--+-- NOTE: The write part of the codecs is largely untested as of yet, as it is+-- not used. It exists to match the tomland interface.+module Hercules.Agent.Config.Json where++import Control.Arrow (left)+import Control.Category ((>>>))+import Control.Lens+  ( Iso',+    Prism',+    Traversal',+    at,+    iso,+    re,+    (%~),+    (^.),+    (^?),+  )+import Control.Monad.Trans.Maybe+import Data.Aeson qualified as Aeson+import Data.Aeson.Key qualified as AK+import Data.Aeson.KeyMap qualified as AM+import Data.Aeson.Lens qualified as Aeson.Lens+import Data.Aeson.Text qualified as Aeson+import Data.ByteString.Lazy qualified as BL+import Data.Map qualified as M+import Data.Profunctor (Profunctor (..), Star (..))+import Data.Text qualified as T+import Data.Text.Lazy qualified as TL+import Data.Vector qualified as V+import Numeric.Lens (integral)+import Protolude hiding (to)+import Toml hiding+  ( decode,+    decodeFile,+    dimap,+    first,+    match,+    _EnumBounded,+    _Text,+    _TextBy,+  )++-- start codec++-- | Generic codec type.+data GCodec r w c a = GCodec+  { gRead :: r a,+    -- | `c -> w a`+    gWrite :: Star w c a+  }+  deriving (Functor)++instance (Applicative r, Applicative w) => Applicative (GCodec r w c) where+  pure x = GCodec (pure x) (pure x)+  GCodec r1 w1 <*> GCodec r2 w2 = GCodec (r1 <*> r2) (w1 <*> w2)++instance (Monad r, Monad w) => Monad (GCodec r w c) where+  (GCodec r w) >>= f =+    GCodec+      (r >>= gRead . f)+      (w >>= gWrite . f)++instance (Functor r, Functor w) => Profunctor (GCodec r w) where+  dimap f g (GCodec r w) = GCodec (fmap g r) (Data.Profunctor.dimap f g w)++instance (Alternative r, Alternative w) => Alternative (GCodec r w c) where+  empty = GCodec empty empty+  GCodec r1 w1 <|> GCodec r2 w2 = GCodec (r1 <|> r2) (w1 <|> w2)++-- end codec++type JsonEnv = ExceptT Text (Reader Aeson.Value)++type JsonSt = MaybeT (State Aeson.Value)++type JsonCodec = GCodec JsonEnv JsonSt++type JsonCodec' a = GCodec JsonEnv JsonSt a a++type JsonBiMap = BiMap Text++(.=.) :: (Profunctor p) => p field a -> (object -> field) -> p object a+a .=. f = dimap f identity a++proOptional :: (Alternative r, Applicative w) => GCodec r w c a -> GCodec r w (Maybe c) (Maybe a)+proOptional (GCodec rd wr) = GCodec (optional rd) (Star $ traverse (runStar wr))++-- | Decode a value from a file. In case of parse errors, throws 'LoadTomlException'.+decodeFile :: (MonadIO m) => JsonCodec' a -> FilePath -> m a+decodeFile codec filePath =+  liftIO $ (decode codec <$> BL.readFile filePath) >>= errorWhenLeft+  where+    errorWhenLeft :: Either Text a -> IO a+    errorWhenLeft (Left e) =+      throwIO $ FatalError $ "In JSON file " <> show filePath <> ": " <> e+    errorWhenLeft (Right pc) = pure pc++-- | Convert textual representation of JSON into user data type.+decode :: JsonCodec' a -> BL.ByteString -> Either Text a+decode codec txt = do+  toml <- first toS $ Aeson.eitherDecode txt+  runCodec codec toml++-- | Convert JSON into user data type.+runCodec :: JsonCodec' a -> Aeson.Value -> Either Text a+runCodec codec = runReader (runExceptT $ gRead codec)++-- | Convert JSON to textual representation.+encode :: JsonCodec' a -> a -> TL.Text+encode codec obj = Aeson.encodeToLazyText $ execCodec codec obj++-- | Runs 'codecWrite' of 'JsonCodec' and returns intermediate Aeson AST.+execCodec :: JsonCodec' a -> a -> Aeson.Value+execCodec codec obj =+  execState (runMaybeT $ runStar (gWrite codec) obj) (Aeson.Object mempty)++at' :: Key -> Traversal' Aeson.Value Aeson.Value+at' (Key ks) = foldr (\(Piece p) r -> Aeson.Lens.key (AK.fromText p) . r) identity ks++insertKeyValue :: Key -> Aeson.Value -> Aeson.Value -> Aeson.Value+insertKeyValue (Key ks) v =+  foldr+    (\(Piece p) r -> Aeson.Lens._Object . at (AK.fromText p) %~ s r)+    (const v)+    ks+  where+    s :: (Aeson.Value -> Aeson.Value) -> Maybe Aeson.Value -> Maybe Aeson.Value+    s setSub (Just x) = Just (setSub x)+    s setSub Nothing = Just (setSub (Aeson.Object mempty))++match :: forall a. JsonBiMap a Aeson.Value -> Key -> JsonCodec' a+match bm key = GCodec input (Star output)+  where+    input :: JsonEnv a+    input = do+      mVal <- asks $ (^? at' key)+      case mVal of+        Nothing -> throwError $ "Not found: " <> showKey key+        Just val -> case backward bm val of+          Right v -> pure v+          Left err -> throwError $ "In " <> showKey key <> ": " <> err++    output :: a -> JsonSt a+    output a = do+      val <- MaybeT $ pure $ either (const Nothing) Just $ forward bm a+      a <$ modify (insertKeyValue key val)++showKey :: Key -> Text+showKey (Key ps) = T.intercalate "." (Protolude.map (toS . unPiece) (Protolude.toList ps))++text :: Key -> JsonCodec' Text+text = match $ prismWithError "String expected" Aeson.Lens._String++string :: Key -> JsonCodec' [Char]+string = match $ prismWithError "String expected" (Aeson.Lens._String . toS')++toS' :: Control.Lens.Iso' Text [Char]+toS' = Control.Lens.iso toS toS++integer :: Key -> JsonCodec' Integer+integer = match $ prismWithError "Integer expected" Aeson.Lens._Integer++int :: Key -> JsonCodec' Int+int = match $ prismWithError "Int expected" $ Aeson.Lens._Integer . integral++bool :: Key -> JsonCodec' Bool+bool = match $ prismWithError "Boolean expected" Aeson.Lens._Bool++value :: Key -> JsonCodec' Aeson.Value+value = match $ prismWithError "Value expected" identity++value' :: JsonCodec' Aeson.Value+value' = GCodec ask (Star (\x -> put x >> pure x))++dimatch ::+  (b -> Maybe a) ->+  (a -> b) ->+  JsonCodec' a ->+  JsonCodec' b+dimatch match_ make inner =+  GCodec+    { gRead = fmap make (gRead inner),+      gWrite = Star $ \b -> do+        case match_ b of+          Just a -> make <$> (runStar (gWrite inner) a)+          Nothing -> empty+    }++emap :: (e -> e2) -> BiMap e a b -> BiMap e2 a b+emap f (BiMap forward_ backward_) = BiMap (left f . forward_) (left f . backward_)++_EnumBounded :: (Show a, Enum a, Bounded a) => JsonBiMap a Aeson.Value+_EnumBounded = emap Toml.prettyBiMapError _EnumBoundedText >>> prismWithError "String expected" Aeson.Lens._String++_KeyText :: BiMap Text Key Text+_KeyText = emap Toml.prettyBiMapError Toml._KeyText++_Text :: BiMap Text Text Aeson.Value+_Text = prismWithError "String expected" Aeson.Lens._String++textBy :: (a -> Text) -> (Text -> Either Text a) -> Key -> JsonCodec' a+textBy to from_ = match (_TextBy to from_)++_TextBy ::+  forall a.+  (a -> Text) ->+  (Text -> Either Text a) ->+  BiMap Text a Aeson.Value+_TextBy toText parseText = BiMap toAnyValue fromAnyValue+  where+    toAnyValue :: a -> Either Text Aeson.Value+    toAnyValue = Right . Aeson.String . toText++    fromAnyValue :: Aeson.Value -> Either Text a+    fromAnyValue (Aeson.String v) = parseText v+    fromAnyValue _ = Left "String expected"++enumBounded :: (Bounded a, Enum a, Show a) => Key -> JsonCodec' a+enumBounded key = match (_EnumBounded) key++arrayOf :: forall a. BiMap Text a Aeson.Value -> Key -> JsonCodec' [a]+arrayOf m k = GCodec input (Star output)+  where+    input :: JsonEnv [a]+    input = do+      mVal <- asks $ (^? at' k)+      case mVal of+        Nothing -> throwError $ "Not found: " <> showKey k+        Just (Aeson.Array arr) -> do+          let go = traverse (backward m)+          case go (Protolude.toList arr) of+            Right v -> pure v+            Left err -> throwError $ "In " <> showKey k <> ": " <> err+        Just _ -> throwError $ "Array expected in " <> showKey k++    output :: [a] -> JsonSt [a]+    output as = do+      as' <- for as $ \a -> do+        MaybeT $ pure $ either (const Nothing) Just $ forward m a+      let arr = Aeson.Array . V.fromList $ as'+      modify (insertKeyValue k arr)+      pure as++tableMap' ::+  JsonCodec' v ->+  Key ->+  JsonCodec' (Map Text v)+tableMap' valCodec key =+  let c = match (prismWithError "JSON Object expected" (Aeson.Lens._Object . aesonMap)) key+   in GCodec+        { gRead = do+            x <- gRead c+            fmap M.fromList $ for (M.toList x) $ \(k, v) -> do+              v' <- local (const v) $ gRead valCodec+              pure (k, v'),+          gWrite = do+            panic "tableMap': write not implemented"+            -- for_ (M.toList m) $ \(k, v) -> do+            --   restore <- get+            --   put $ Aeson.object []+            --   _v <- runStar (gWrite valCodec) v+            --   v' <- get+            --   put restore+            --   modify (insertKeyValue k' v')+            -- pure _+        }++aesonMap :: Control.Lens.Iso' (AM.KeyMap a) (Map Text a)+aesonMap = Control.Lens.iso AM.toMapText AM.fromMapText++aesonMap' :: Control.Lens.Iso' (Map Text a) (AM.KeyMap a)+aesonMap' = Control.Lens.iso AM.fromMapText AM.toMapText++prismWithError :: e -> Prism' a b -> BiMap e b a+prismWithError e p =+  BiMap+    { forward = \x -> Right (x ^. re p),+      backward = \y -> maybeToEither e $ y ^? p+    }

@@ -0,0 +1,58 @@+{-# LANGUAGE GADTs #-}+{-# OPTIONS_GHC -Wno-orphans #-}++module Hercules.Agent.Config.Toml where++import Data.Aeson qualified as A+import Data.Aeson.KeyMap qualified as AK+import Data.Profunctor (Profunctor (..))+import Data.Scientific (floatingOrInteger, fromFloatDigits)+import Data.Vector qualified as V+import Protolude hiding (to)+import Toml++instance Profunctor Toml.Codec where+  dimap f g (Toml.Codec r w) = Toml.Codec (\x -> g <$> r x) (\a -> g <$> (w (f a)))++embedJson :: Key -> TomlCodec A.Value+embedJson key =+  Codec+    { codecRead =+        codecRead (Toml.match (embedJsonBiMap key) key)+          <!> codecRead (A.Object . AK.fromHashMapText <$> Toml.tableHashMap Toml._KeyText embedJson key),+      codecWrite = panic "embedJson.write: not implemented" $ \case+        A.String s -> A.String <$> codecWrite (Toml.text key) s+        A.Number sci -> A.Number . fromRational . toRational <$> codecWrite (Toml.double key) (fromRational $ toRational sci)+        A.Bool b -> A.Bool <$> codecWrite (Toml.bool key) b+        A.Array a -> A.Array . V.fromList <$> codecWrite (Toml.arrayOf (embedJsonBiMap key) key) (Protolude.toList a)+        A.Object o -> A.Object . AK.fromHashMapText <$> codecWrite (Toml.tableHashMap Toml._KeyText embedJson key) (AK.toHashMapText o)+        A.Null -> eitherToTomlState (Left ("null is not supported in TOML" :: Text))+    }++embedJsonBiMap :: Key -> TomlBiMap A.Value AnyValue+embedJsonBiMap _key =+  -- TODO: use key for error reporting+  BiMap+    { forward = panic "embedJsonBiMap.forward: not implemented" $ \case+        A.String s -> pure $ AnyValue $ Text s+        A.Number sci -> case floatingOrInteger sci of+          Left fl -> pure $ AnyValue $ Double fl -- lossy+          Right i -> pure $ AnyValue $ Integer i+        A.Bool b -> pure $ AnyValue $ Bool b+        A.Array _a -> Left $ ArbitraryError "Conversion from JSON array of arrays to TOML not implemented yet"+        A.Object _o -> Left $ ArbitraryError "Conversion from JSON array of objects to TOML is not supported"+        A.Null -> Left $ ArbitraryError "JSON null is not supported in TOML",+      backward = anyValueToJSON+    }++anyValueToJSON :: AnyValue -> Either TomlBiMapError A.Value+anyValueToJSON = \case+  AnyValue (Bool b) -> pure (A.Bool b)+  AnyValue (Integer i) -> pure (A.Number $ fromIntegral i)+  AnyValue (Double d) -> pure (A.Number $ fromFloatDigits d)+  AnyValue (Text t) -> pure (A.String t)+  AnyValue (Zoned _zt) -> Left (ArbitraryError "Conversion from TOML zoned time to JSON not implemented yet. Use a string.")+  AnyValue (Local _zt) -> Left (ArbitraryError "Conversion from TOML local time to JSON not implemented yet. Use a string.")+  AnyValue (Day _d) -> Left (ArbitraryError "Conversion from TOML day to JSON not implemented yet. Use a string.")+  AnyValue (Hours _h) -> Left (ArbitraryError "Conversion from TOML hours to JSON not implemented yet. Use a string.")+  AnyValue (Array a) -> A.Array <$> sequence (V.fromList (a <&> AnyValue <&> anyValueToJSON))

@@ -3,35 +3,34 @@ import Data.Aeson qualified as A import Data.IORef import Data.Map qualified as M+import Data.Vector (Vector) import Hercules.API.Agent.Effect.EffectTask qualified as EffectTask+import Hercules.API.Logs.LogEntry (LogEntry) import Hercules.API.TaskStatus (TaskStatus) import Hercules.API.TaskStatus qualified as TaskStatus import Hercules.Agent.Config qualified as Config import Hercules.Agent.Env hiding (config) import Hercules.Agent.Env qualified as Env import Hercules.Agent.Files+import Hercules.Agent.InitWorkerConfig qualified as InitWorkerConfig import Hercules.Agent.Log-import Hercules.Agent.Nix qualified as Nix import Hercules.Agent.Sensitive (Sensitive (Sensitive))-import Hercules.Agent.ServiceInfo qualified as ServiceInfo import Hercules.Agent.WorkerProcess import Hercules.Agent.WorkerProcess qualified as WorkerProcess import Hercules.Agent.WorkerProtocol.Command qualified as Command import Hercules.Agent.WorkerProtocol.Command.Effect qualified as Command.Effect import Hercules.Agent.WorkerProtocol.Event qualified as Event-import Hercules.Agent.WorkerProtocol.LogSettings qualified as LogSettings import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON (ViaJSON)) import Hercules.Secrets qualified as Secrets-import Network.URI qualified import Protolude import System.Posix.Signals qualified as PS import System.Process -performEffect :: EffectTask.EffectTask -> App TaskStatus-performEffect effectTask = withWorkDir "effect" $ \workDir -> do+performEffect :: (Vector LogEntry -> IO ()) -> EffectTask.EffectTask -> App TaskStatus+performEffect sendLogEntries effectTask = withWorkDir "effect" $ \workDir -> do   workerExe <- getWorkerExe   commandChan <- liftIO newChan-  extraNixOptions <- Nix.askExtraOptions+  workerConfig <- InitWorkerConfig.getWorkerConfig   workerEnv <-     liftIO $       WorkerProcess.prepareEnv@@ -41,23 +40,22 @@             }         )   effectResult <- liftIO $ newIORef Nothing-  let opts = [show extraNixOptions]-      procSpec =-        (System.Process.proc workerExe opts)+  let procSpec =+        (System.Process.proc workerExe ["effect", toS effectTask.derivationPath])           { env = Just workerEnv,             close_fds = True,             cwd = Just workDir           }       writeEvent :: Event.Event -> App ()       writeEvent event = case event of+        Event.LogItems (ViaJSON e) -> do+          liftIO (sendLogEntries e)         Event.EffectResult e -> do           liftIO $ writeIORef effectResult (Just e)         Event.Exception e -> do           panic e         _ -> pass   config <- asks Env.config-  let materialize = not (Config.nixUserIsTrusted config)-  baseURL <- asks (ServiceInfo.bulkSocketBaseURL . Env.serviceInfo)   liftIO $     writeChan commandChan $       Just $@@ -65,13 +63,6 @@           Command.Effect.Effect             { drvPath = EffectTask.derivationPath effectTask,               inputDerivationOutputPaths = encodeUtf8 <$> EffectTask.inputDerivationOutputPaths effectTask,-              logSettings =-                LogSettings.LogSettings-                  { token = Sensitive $ EffectTask.logToken effectTask,-                    path = "/api/v1/logs/build/socket",-                    baseURL = toS $ Network.URI.uriToString identity baseURL ""-                  },-              materializeDerivation = materialize,               secretsPath = toS $ Config.secretsJsonPath config,               serverSecrets = Sensitive $ ViaJSON (EffectTask.serverSecrets effectTask),               token = Sensitive (EffectTask.token effectTask),@@ -84,17 +75,19 @@                     repoName = EffectTask.repoName effectTask,                     ref = EffectTask.ref effectTask,                     isDefaultBranch = EffectTask.isDefaultBranch effectTask-                  }+                  },+              configuredMountables = Sensitive (ViaJSON (Config.effectMountables config))             }   let stderrHandler =         stderrLineHandler+          sendLogEntries           ( M.fromList               [ ("taskId", A.toJSON (EffectTask.id effectTask)),                 ("derivationPath", A.toJSON (EffectTask.derivationPath effectTask))               ]           )           "Effect worker"-  exitCode <- runWorker procSpec stderrHandler commandChan writeEvent+  exitCode <- runWorker workerConfig 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)"

@@ -16,10 +16,12 @@   ) import Hercules.Agent.Config (FinalConfig) import Hercules.Agent.Config.BinaryCaches qualified as Config.BinaryCaches+import Hercules.Agent.Memo (Memo) import Hercules.Agent.Netrc.Env qualified as Netrc import Hercules.Agent.Nix.Env qualified as Nix   ( Env,   )+import Hercules.Agent.ResourceLimiter (ResourceLimiter) import Hercules.Agent.ServiceInfo qualified as ServiceInfo import Hercules.Agent.Socket (Socket) import Hercules.Error@@ -48,7 +50,16 @@     -- katip     kNamespace :: K.Namespace,     kContext :: K.LogContexts,-    kLogEnv :: K.LogEnv+    kLogEnv :: K.LogEnv,+    -- | Limits concurrent store operations during evaluation. For use with caches.+    --+    -- May overlap with 'concurrentStorePushes' in practice, but the excess+    -- shouldn't be a problem. By having two counters we avoid starving either task.+    concurrentStoreQueries :: Memo Text ResourceLimiter,+    -- | Limits concurrent store operations during evaluation. For use with caches.+    --+    -- See 'concurrentStoreQueries'.+    concurrentStorePushes :: Memo Text ResourceLimiter   }  activePushCaches :: App [Text]

@@ -13,6 +13,7 @@ import Hercules.Agent.Log import Hercules.CNix qualified as CNix import Hercules.CNix.Settings qualified as Settings+import Hercules.CNix.Store (withStoreFromURI) import Hercules.CNix.Store qualified as Store import Network.HostName (getHostName) import Protolude hiding (to)@@ -33,7 +34,7 @@             nixVersion = nixLibVersion nix,             nixClientProtocolVersion = nixClientProtocolVersion,             nixDaemonProtocolVersion = nixStoreProtocolVersion,-            platforms = map fromUtf8Lenient $ nixPlatforms nix,+            platforms = map fromUtf8Lenient (nixPlatforms nix) <> fromMaybe [] cfg.remotePlatformsWithSameFeatures,             cachixPushCaches = cachixPushCaches,             pushCaches = pushCaches,             systemFeatures = map fromUtf8Lenient $ nixSystemFeatures nix,@@ -67,17 +68,19 @@   trustedPublicKeys <- Settings.getTrustedPublicKeys   narinfoCacheNegativeTTL <- Settings.getNarinfoCacheNegativeTtl   netrcFile <- Settings.getNetrcFile+  cleanSubstituters <- ordNub <$> traverse cleanUrl substituters   pure     NixInfo       { nixLibVersion = T.dropAround isSpace (fromUtf8Lenient CNix.nixVersion),         nixPlatforms = toList (S.singleton system <> extraPlatforms),         nixSystemFeatures = toList systemFeatures,-        nixSubstituters = map cleanUrl substituters,+        nixSubstituters = cleanSubstituters,         nixTrustedPublicKeys = trustedPublicKeys,         nixNarinfoCacheNegativeTTL = narinfoCacheNegativeTTL,         nixNetrcFile = guard (netrcFile /= "") $> netrcFile       } -cleanUrl :: ByteString -> ByteString-cleanUrl t | "@" `BS.isInfixOf` t = "<URI censored; might contain secret>"-cleanUrl t = t+cleanUrl :: ByteString -> IO ByteString+cleanUrl t | "@" `BS.isInfixOf` t = pure "<URI censored; might contain secret>"+cleanUrl t = do+  withStoreFromURI (decodeUtf8With lenientDecode t) Store.storeUri

@@ -8,9 +8,10 @@ where  import Conduit+import Control.Concurrent.Async (waitSTM) import Control.Concurrent.Async.Lifted qualified as Async.Lifted import Control.Concurrent.Chan.Lifted-import Control.Exception.Lifted (finally)+import Control.Concurrent.STM.TVar (TVar, newTVarIO) import Control.Exception.Safe qualified as Safe import Control.Lens (at, (^?)) import Control.Monad.IO.Unlift (askUnliftIO, unliftIO)@@ -22,12 +23,11 @@ import Data.IORef   ( atomicModifyIORef,     newIORef,-    readIORef,   ) import Data.Map qualified as M-import Data.Set qualified as S import Data.Text qualified as T import Data.UUID (UUID)+import Data.Vector (Vector) import Hercules.API (Id) import Hercules.API.Agent.Evaluate   ( getDerivationStatus2,@@ -44,22 +44,25 @@ import Hercules.API.Agent.Evaluate.EvaluateEvent.DerivationInfo qualified as DerivationInfo import Hercules.API.Agent.Evaluate.EvaluateEvent.JobConfig qualified as JobConfig import Hercules.API.Agent.Evaluate.EvaluateEvent.Message qualified as Message-import Hercules.API.Agent.Evaluate.EvaluateEvent.PushedAll qualified as PushedAll+import Hercules.API.Agent.Evaluate.EvaluateEvent.SubstitutionQueryResult qualified as SubstitutionQueryResult import Hercules.API.Agent.Evaluate.EvaluateTask qualified as EvaluateTask import Hercules.API.Agent.Evaluate.ImmutableGitInput (ImmutableGitInput) import Hercules.API.Agent.Evaluate.ImmutableGitInput qualified as ImmutableGitInput import Hercules.API.Agent.Evaluate.ImmutableInput qualified as ImmutableInput+import Hercules.API.Logs.LogEntry (LogEntry) import Hercules.API.Servant (noContent) import Hercules.API.Task (Task)+import Hercules.Agent.Build (convertOutputInfo)+import Hercules.Agent.Cache (getConfiguredSubstituters) import Hercules.Agent.Cache qualified as Agent.Cache import Hercules.Agent.Client qualified import Hercules.Agent.Config qualified as Config import Hercules.Agent.Env import Hercules.Agent.Env qualified as Env-import Hercules.Agent.Evaluate.TraversalQueue (Queue)-import Hercules.Agent.Evaluate.TraversalQueue qualified as TraversalQueue import Hercules.Agent.Files+import Hercules.Agent.InitWorkerConfig qualified as WorkerConfig import Hercules.Agent.Log+import Hercules.Agent.Memo (Memo, doOnce, newMemo) import Hercules.Agent.Netrc qualified as Netrc import Hercules.Agent.Nix qualified as Nix import Hercules.Agent.Nix.RetrieveDerivationInfo@@ -71,9 +74,8 @@   ( renderSubPath,   ) import Hercules.Agent.Producer-import Hercules.Agent.Sensitive (Sensitive (Sensitive))-import Hercules.Agent.ServiceInfo qualified as ServiceInfo-import Hercules.Agent.WorkerProcess ()+import Hercules.Agent.ResourceLimiter (ResourceLimiter, newResourceLimiter, withResource)+import Hercules.Agent.Store (toDrvInfo) import Hercules.Agent.WorkerProcess qualified as WorkerProcess import Hercules.Agent.WorkerProtocol.Command qualified as Command import Hercules.Agent.WorkerProtocol.Command.BuildResult qualified as BuildResult@@ -82,22 +84,23 @@ import Hercules.Agent.WorkerProtocol.Event.Attribute qualified as WorkerAttribute import Hercules.Agent.WorkerProtocol.Event.AttributeError qualified as WorkerAttributeError import Hercules.Agent.WorkerProtocol.Event.AttributeIFD qualified as AttributeIFD-import Hercules.Agent.WorkerProtocol.LogSettings qualified as LogSettings-import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON (ViaJSON), fromViaJSON)-import Hercules.CNix.Store (Store, StorePath, parseStorePath)+import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON (ViaJSON))+import Hercules.Agent.WorkerProtocol.ViaJSON qualified+import Hercules.CNix.Store (Store, StorePath, getStorePathBaseName, parseStorePath) import Hercules.CNix.Store qualified as CNix import Hercules.Effect (parseDrvSecretsMap) import Hercules.Error (defaultRetry, quickRetry) import Network.HTTP.Client.Conduit qualified as HTTP.Conduit import Network.HTTP.Simple qualified as HTTP.Simple import Network.URI qualified-import Protolude hiding (finally, newChan, writeChan)+import Protolude hiding (async, atomically, concurrently, finally, newChan, writeChan) import Servant.Client qualified import Servant.Client.Core (showBaseUrl) import System.Directory qualified as Dir import System.FilePath import System.Process-import UnliftIO (atomicModifyIORef')+import UnliftIO (async, atomicModifyIORef', atomically, concurrently, concurrently_, forConcurrently, forConcurrently_, modifyTVar, readTVar, writeTVar)+import UnliftIO qualified  eventLimit :: Int eventLimit = 50000@@ -105,9 +108,9 @@ pushEvalWorkers :: Int pushEvalWorkers = 16 -performEvaluation :: Store -> EvaluateTask.EvaluateTask -> App ()-performEvaluation store task' =-  withProducer (produceEvaluationTaskEvents store task') $ \producer ->+performEvaluation :: (Vector LogEntry -> IO ()) -> Store -> EvaluateTask.EvaluateTask -> App ()+performEvaluation sendLogItems store task' =+  withProducer (produceEvaluationTaskEvents sendLogItems store task') $ \producer ->     withBoundedDelayBatchProducer (1000 * 1000) 1000 producer $ \batchProducer ->       fix $ \continue ->         joinSTM $ listen batchProducer (\b -> withSync b (postBatch task' . catMaybes) *> continue) pure@@ -121,6 +124,27 @@   Just {} -> do     pure Nothing +data AbortMessageAlreadySent = AbortMessageAlreadySent deriving (Show, Exception)++getWithStoreLimiter :: (MonadUnliftIO m) => (Env -> Memo Text ResourceLimiter) -> Text -> App (m a -> m a)+getWithStoreLimiter getLimiterMap store = do+  memo <- asks getLimiterMap+  limiter <- doOnce memo store do+    newResourceLimiter 32+  pure (withResource limiter)++-- | Apply concurreny limit+withStoreQuery :: Text -> App a -> App a+withStoreQuery storeURI m = do+  withLimit <- getWithStoreLimiter Env.concurrentStoreQueries storeURI+  withLimit m++-- | Apply concurreny limit+withStorePush :: Text -> App a -> App a+withStorePush storeURI m = do+  withLimit <- getWithStoreLimiter Env.concurrentStorePushes storeURI+  withLimit m+ makeEventEmitter ::   (Syncing EvaluateEvent.EvaluateEvent -> App ()) ->   App (EvaluateEvent.EvaluateEvent -> App ())@@ -171,218 +195,374 @@         emitSingle =<< fixIndex update   pure emit +-- | @withDynamicBarrier (\addWait -> m ...)@ waits for any number of concurrent operations,+-- which are registered during the execution of @m@ using @addWait@.+--+-- The @addWait@ function enqueues an 'STM' transaction that will be dequeued only when the transaction completes successfully.+withDynamicBarrier :: (MonadUnliftIO m) => ((STM x -> STM ()) -> m a) -> m a+withDynamicBarrier driver = do+  -- Signals that `driver` is done. Prevents exiting before work builds up.+  driverDone :: TVar (STM ()) <- liftIO (newTVarIO retry)++  -- Collection of work that needs to complete before returning, including the `driver` itself.+  -- It's a list so that wait conditions that have completed can be removed; not checked again.+  done :: TVar [STM ()] <- liftIO (newTVarIO [join $ readTVar driverDone])++  let blockOn :: STM a -> STM ()+      blockOn stm = do+        modifyTVar done (void stm :)++      waitDone =+        atomically isDone >>= \case+          True -> pass+          False -> waitDone++      isDone :: STM Bool+      isDone = do+        readTVar done >>= \case+          [] -> pure True+          (c : cs) -> do+            c+            writeTVar done cs+            pure False++      claimDriverDone = writeTVar driverDone pass++  fst <$> concurrently (driver blockOn `UnliftIO.finally` atomically claimDriverDone) waitDone++withStores :: [Text] -> (Map Text Store -> App r) -> App r+withStores storeURIs f =+  let sorted = ordNub storeURIs+   in foldr+        ( \uri f2 stores -> do+            CNix.withStoreFromURI uri \store ->+              f2 (stores <> M.singleton uri store)+        )+        f+        sorted+        mempty++withSubstituters :: (Map Text Store -> App r) -> App r+withSubstituters f = do+  substituterURIs <- getConfiguredSubstituters+  withStores substituterURIs f+ produceEvaluationTaskEvents ::+  (Vector LogEntry -> IO ()) ->   Store ->   EvaluateTask.EvaluateTask ->   (Syncing EvaluateEvent.EvaluateEvent -> App ()) ->   App ()-produceEvaluationTaskEvents store task writeToBatch | EvaluateTask.isFlakeJob task = withWorkDir "eval" $ \tmpdir -> do-  logLocM DebugS "Retrieving evaluation task (flake)"+produceEvaluationTaskEvents sendLogItems store task writeToBatch = UnliftIO.handle (\AbortMessageAlreadySent -> pass) $ withSubstituters \substituters -> withWorkDir "eval" $ \tmpdir -> do   let sync = syncer writeToBatch-  topDerivationPaths <- liftIO $ newIORef mempty   emit <- makeEventEmitter writeToBatch-  let allowedPaths = [] -  TraversalQueue.with $ \(derivationQueue :: Queue StorePath) ->-    let doIt =-          ( do-              Async.Lifted.concurrently_ evaluation emitDrvs-              -- derivationInfo upload has finished-              -- allAttrPaths :: IORef has been populated+  derivationInfoUpload :: Memo ByteString () <- newMemo+  derivationSubstitutable :: Memo (Text, ByteString) Bool <- newMemo+  derivationBuildRequest :: Memo Text () <- newMemo+  derivationCache :: Memo ByteString CNix.Derivation <- newMemo+  planBuild_ <- newMemo++  let emitDrvInfoRaw :: StorePath -> App ()+      emitDrvInfoRaw drvPath = do+        drvInfo <- retrieveDerivationInfo store drvPath+        forConcurrently_ (M.keys $ DerivationInfo.inputDerivations drvInfo) \inp -> do+          inputStorePath <- liftIO $ parseStorePath store (encodeUtf8 inp)+          emitDrvInfo inputStorePath+        emit $ EvaluateEvent.DerivationInfo drvInfo++      getDerivationCached :: StorePath -> App CNix.Derivation+      getDerivationCached drvPath = do+        bs <- liftIO do getStorePathBaseName drvPath+        doOnce derivationCache bs do+          liftIO $ CNix.getDerivation store drvPath++      emitDrvInfo :: StorePath -> App ()+      emitDrvInfo sp = do+        bs <- liftIO (CNix.getStorePathBaseName sp)+        doOnce derivationInfoUpload bs do+          emitDrvInfoRaw sp++      rawQuerySubstitutableOutput :: StorePath -> CNix.Derivation -> Text -> ByteString -> App Bool+      rawQuerySubstitutableOutput drvPath drv drvPathText outputName = do+        drvName <- liftIO $ CNix.getDerivationNameFromPath drvPath+        outputs <- liftIO $ CNix.getDerivationOutputs store drvName drv+        output <- case find (\o -> o.derivationOutputName == outputName) outputs of+          Nothing -> panic $ "derivation " <> drvPathText <> " does not have output " <> show outputName+          Just x -> pure x+        querySubstitutableOutput' drvPathText outputName output++      querySubstitutableOutput' drvPathText outputName output = do+        doOnce derivationSubstitutable (drvPathText, outputName) do+          rawQuerySubstitutableOutput' drvPathText outputName output++      rawQuerySubstitutableOutput' drvPathText outputName output = do+        outputPath <- case output.derivationOutputPath of+          Nothing -> panic $ "derivation " <> drvPathText <> " does not have a predetermined path. Content addressed derivations are not supported yet."+          Just x -> pure x+        alreadyPositive <-+          any (isJust . join) <$> for (M.toList substituters) \(_uri, substituter) -> liftIO do+            CNix.queryPathInfoFromClientCache substituter outputPath+        if alreadyPositive+          then pure True+          else do+            -- TODO waitAny+            any identity <$> forConcurrently (M.toList substituters) \(uri, substituter) -> do+              exists <- withStoreQuery uri do+                liftIO (CNix.isValidPath substituter outputPath)+              if exists+                then do+                  drvInfo <- liftIO (toDrvInfo substituter output)+                  emit $+                    EvaluateEvent.SubstitutionQueryResult+                      SubstitutionQueryResult.SubstitutionQueryResult+                        { storeURI = uri,+                          derivation = drvPathText,+                          outputName = decodeUtf8With lenientDecode outputName,+                          outputInfo = Just (convertOutputInfo drvPathText drvInfo)+                        }+                else do+                  emit $+                    EvaluateEvent.SubstitutionQueryResult+                      SubstitutionQueryResult.SubstitutionQueryResult+                        { storeURI = uri,+                          derivation = drvPathText,+                          outputName = decodeUtf8With lenientDecode outputName,+                          outputInfo = Nothing+                        }++              pure exists++      storePathToText :: StorePath -> App Text+      storePathToText sp = do+        liftIO (CNix.storePathToPath store sp) <&> decodeUtf8With lenientDecode++      planAllOutputs :: StorePath -> App ()+      planAllOutputs drvPath = do+        drvPathText <- storePathToText drvPath+        drv <- getDerivationCached drvPath+        drvName <- liftIO $ CNix.getDerivationNameFromPath drvPath+        outputs <- liftIO $ CNix.getDerivationOutputs store drvName drv+        outputsSubstitutable <-+          all identity <$> forConcurrently outputs \output ->+            querySubstitutableOutput' drvPathText output.derivationOutputName output++        when (not outputsSubstitutable) do+          planBuild drv drvPathText++      -- Like 'planInputs' but do assume we need to build all of them.+      planInputsForced drvPath = do+        drv <- getDerivationCached drvPath+        inputs <- liftIO (CNix.getDerivationInputs' drv)+        for_ inputs \(inputDrvPath, _outputs) -> do+          inputDrv <- getDerivationCached inputDrvPath+          planBuild inputDrv =<< storePathToText inputDrvPath++      planInputs drv = do+        inputs <- liftIO (CNix.getDerivationInputs' drv)+        for_ inputs \(inputDrvPath, outputs) -> do+          inputDrvPathText <- decodeUtf8With lenientDecode <$> liftIO (CNix.storePathToPath store inputDrvPath)+          inputDrv <- getDerivationCached inputDrvPath+          for outputs \outputName -> do+            planOutput inputDrvPath inputDrv outputName inputDrvPathText++      planOutput :: StorePath -> CNix.Derivation -> ByteString -> Text -> App ()+      planOutput drvPath drv outputName drvPathText = do+        isSubstitutable <- rawQuerySubstitutableOutput drvPath drv drvPathText outputName+        if isSubstitutable+          then pass+          else do+            planBuild drv drvPathText++      -- Plan a build, when it has been determined that a build task is necessary.+      planBuild drv drvPathText =+        doOnce planBuild_ drvPathText do+          planInputs drv+          requestBuild drvPathText++      requestBuild drvPathText =+        doOnce derivationBuildRequest drvPathText do+          emit $ EvaluateEvent.BuildRequest BuildRequest.BuildRequest {derivationPath = drvPathText, forceRebuild = False}++  let isFlakeJob = EvaluateTask.isFlakeJob task+  nonFlakeStuff <- for (guard (not isFlakeJob)) \() -> do+    inputLocations <-+      EvaluateTask.otherInputs task+        & M.traverseWithKey \k src -> do+          let fetchName = tmpdir </> ("fetch-" <> toS k)+              argName = tmpdir </> ("arg-" <> toS k)+              metaName = do+                meta <- EvaluateTask.inputMetadata task & M.lookup k+                nameValue <- meta & M.lookup "name"+                name <- case A.fromJSON nameValue of+                  A.Success a -> pure a+                  _ -> Nothing+                guard (isValidName name)+                pure name+              destName+                | Just sourceName <- metaName = argName </> sourceName+                | otherwise = argName+          fetched <- fetchSource fetchName src+          liftIO do+            Dir.createDirectoryIfMissing True (takeDirectory destName)+            renamePathTryHarder fetched destName+          pure destName+    projectDir <- case M.lookup "src" inputLocations of+      Nothing -> panic "No primary source provided"+      Just x -> pure x+    nixPath <-+      EvaluateTask.nixPath task+        & ( traverse+              . traverse+              . traverse+              $ \identifier -> case M.lookup identifier inputLocations of+                Just x -> pure x+                Nothing ->+                  throwIO $+                    FatalError $+                      "Nix path references undefined input "+                        <> identifier           )-            `finally` pushDrvs+    autoArguments' <-+      EvaluateTask.autoArguments task+        & (traverse . traverse)+          ( \identifier -> case M.lookup identifier inputLocations of+              Just x | "/" `isPrefixOf` x -> pure x+              Just x ->+                throwIO $+                  FatalError $+                    "input "+                      <> identifier+                      <> " was not resolved to an absolute path: "+                      <> toS x+              Nothing ->+                throwIO $+                  FatalError $+                    "auto argument references undefined input "+                      <> identifier+          )+    let autoArguments =+          autoArguments'+            & M.mapWithKey \k sp ->+              let argPath = encodeUtf8 $ renderSubPath $ toS <$> sp+               in case do+                    inputId <- EvaluateTask.autoArguments task & M.lookup k+                    EvaluateTask.inputMetadata task & M.lookup (EvaluateTask.path inputId) of+                    Nothing -> Eval.ExprArg argPath+                    Just attrs ->+                      Eval.ExprArg $+                        -- TODO pass directly to avoid having to escape (or just escape properly)+                        "builtins.fromJSON ''" <> BL.toStrict (A.encode attrs) <> "'' // { outPath = " <> argPath <> "; }"+    adHocSystem <-+      readFileMaybe (projectDir </> "ci-default-system.txt")++    file <-+      liftIO (findNixFile projectDir) >>= \case+        Left e -> do+          emit $+            EvaluateEvent.Message+              Message.Message+                { Message.index = -1, -- will be set by emit+                  Message.typ = Message.Error,+                  Message.message = e+                }+          throwIO AbortMessageAlreadySent+        Right file ->+          pure file++    pure (nixPath, autoArguments, inputLocations, adHocSystem, projectDir, file)++  let nixPath = case nonFlakeStuff of+        Nothing -> mempty+        Just (np, _, _, _, _, _) -> np+      autoArgs = case nonFlakeStuff of+        Nothing -> mempty+        Just (_, aa, _, _, _, _) -> aa+      allowedPaths = case nonFlakeStuff of+        Nothing -> []+        Just (_, _, inputLocations, _, _, _) -> toList inputLocations <&> toS <&> encodeUtf8+      adHocSystem = case nonFlakeStuff of+        Nothing -> Nothing+        Just (_, _, _, a, _, _) -> a+      projectDir = case nonFlakeStuff of+        Nothing -> tmpdir -- unused+        Just (_, _, _, _, pd, _) -> pd+      file = case nonFlakeStuff of+        Nothing -> "" -- unused+        Just (_, _, _, _, _, file_) -> file_++  let uploadDrvs paths = do+        caches <- activePushCaches+        forM_ caches $ \cache -> do+          withNamedContext "cache" cache $ logLocM DebugS "Pushing derivations"++          -- TODO: Make it fine grained? withStorePush here and now is a limit on concurrent closures.+          withStorePush cache do+            Agent.Cache.push store cache (toList paths) pushEvalWorkers++  withDynamicBarrier \addToWait ->+    let addAsync = addToWait . waitSTM         uploadDrvInfos drvPath = do-          TraversalQueue.enqueue derivationQueue drvPath-          TraversalQueue.waitUntilDone derivationQueue+          emitDrvInfo drvPath+          uploadDrvs [drvPath]         addTopDerivation drvPath = do-          TraversalQueue.enqueue derivationQueue drvPath-          liftIO $ atomicModifyIORef topDerivationPaths ((,()) . S.insert drvPath)+          atomically . addAsync =<< async do+            concurrently_+              (emitDrvInfo drvPath)+              (uploadDrvs [drvPath])+            planAllOutputs drvPath+            -- In theory we should be done now+            -- but we want derivation events for these for the backend, and+            -- potentially for inspecting the nix-support directory (TBD).+            -- FIXME: this realises build inputs even if output was substitutable+            --        (but only for top level derivations)+            planInputs =<< getDerivationCached drvPath+            requestBuild =<< storePathToText drvPath++        addTopDerivationInputs drvPath = do+          atomically . addAsync =<< async do+            concurrently_+              (emitDrvInfo drvPath)+              (uploadDrvs [drvPath])+            -- In theory we should be done (as in addTopDerivation) but we want+            -- derivation events for the inputs, so we force them.+            -- FIXME: this realises build inputs even if output was substitutable+            --        (but only for top level deps-only derivations' inputs)+            planInputsForced drvPath++        extraOpts = [("system", T.strip s) | Just s <- [adHocSystem]]         evaluation = do           let evalProc =-                do runEvalProcess-                  store-                  tmpdir -- unused-                  ""-                  mempty-                  mempty-                  captureAttrDrvAndEmit-                  uploadDrvInfos-                  sync-                  task-                  allowedPaths-          evalProc `finally` do-            -- Always upload drv infos, even in case of a crash in the worker-            TraversalQueue.waitUntilDone derivationQueue-              `finally` TraversalQueue.close derivationQueue-        pushDrvs = do-          caches <- activePushCaches-          paths <- liftIO $ readIORef topDerivationPaths-          forM_ caches $ \cache -> do-            withNamedContext "cache" cache $ logLocM DebugS "Pushing derivations"-            Agent.Cache.push store cache (toList paths) pushEvalWorkers-            emit $ EvaluateEvent.PushedAll $ PushedAll.PushedAll {cache = cache}+                Nix.withExtraOptions extraOpts do+                  runEvalProcess+                    sendLogItems+                    store+                    projectDir+                    file+                    autoArgs+                    nixPath+                    captureAttrDrvAndEmit+                    uploadDrvInfos+                    sync+                    task+                    allowedPaths+          evalProc         captureAttrDrvAndEmit msg = do           case msg of             EvaluateEvent.Attribute ae -> do               storePath <- liftIO $ parseStorePath store (encodeUtf8 $ AttributeEvent.derivationPath ae)-              addTopDerivation storePath+              case ae.typ of+                AttributeEvent.Regular -> addTopDerivation storePath+                AttributeEvent.MustFail -> addTopDerivation storePath+                AttributeEvent.MayFail -> addTopDerivation storePath+                AttributeEvent.DependenciesOnly -> addTopDerivationInputs storePath+                AttributeEvent.Effect -> addTopDerivationInputs storePath             EvaluateEvent.AttributeEffect ae -> do               storePath <- liftIO $ parseStorePath store (encodeUtf8 $ AttributeEffectEvent.derivationPath ae)-              addTopDerivation storePath+              addTopDerivationInputs storePath             _ -> pass           emit msg-        emitDrvs =-          TraversalQueue.work derivationQueue $ \recurse drvPath -> do-            drvInfo <- retrieveDerivationInfo store drvPath-            for_ (M.keys $ DerivationInfo.inputDerivations drvInfo) \inp -> do-              inputStorePath <- liftIO $ parseStorePath store (encodeUtf8 inp)-              recurse inputStorePath -- asynchronously-            emit $ EvaluateEvent.DerivationInfo drvInfo-     in doIt-produceEvaluationTaskEvents store task writeToBatch = withWorkDir "eval" $ \tmpdir -> do-  logLocM DebugS "Retrieving evaluation task"-  let sync = syncer writeToBatch-  inputLocations <--    EvaluateTask.otherInputs task-      & M.traverseWithKey \k src -> do-        let fetchName = tmpdir </> ("fetch-" <> toS k)-            argName = tmpdir </> ("arg-" <> toS k)-            metaName = do-              meta <- EvaluateTask.inputMetadata task & M.lookup k-              nameValue <- meta & M.lookup "name"-              name <- case A.fromJSON nameValue of-                A.Success a -> pure a-                _ -> Nothing-              guard (isValidName name)-              pure name-            destName-              | Just sourceName <- metaName = argName </> sourceName-              | otherwise = argName-        fetched <- fetchSource fetchName src-        liftIO do-          Dir.createDirectoryIfMissing True (takeDirectory destName)-          renamePathTryHarder fetched destName-        pure destName-  projectDir <- case M.lookup "src" inputLocations of-    Nothing -> panic "No primary source provided"-    Just x -> pure x-  nixPath <--    EvaluateTask.nixPath task-      & ( traverse-            . traverse-            . traverse-            $ \identifier -> case M.lookup identifier inputLocations of-              Just x -> pure x-              Nothing ->-                throwIO $-                  FatalError $-                    "Nix path references undefined input "-                      <> identifier-        )-  autoArguments' <--    EvaluateTask.autoArguments task-      & (traverse . traverse)-        ( \identifier -> case M.lookup identifier inputLocations of-            Just x | "/" `isPrefixOf` x -> pure x-            Just x ->-              throwIO $-                FatalError $-                  "input "-                    <> identifier-                    <> " was not resolved to an absolute path: "-                    <> toS x-            Nothing ->-              throwIO $-                FatalError $-                  "auto argument references undefined input "-                    <> identifier-        )-  let autoArguments =-        autoArguments'-          & M.mapWithKey \k sp ->-            let argPath = encodeUtf8 $ renderSubPath $ toS <$> sp-             in case do-                  inputId <- EvaluateTask.autoArguments task & M.lookup k-                  EvaluateTask.inputMetadata task & M.lookup (EvaluateTask.path inputId) of-                  Nothing -> Eval.ExprArg argPath-                  Just attrs ->-                    Eval.ExprArg $-                      -- TODO pass directly to avoid having to escape (or just escape properly)-                      "builtins.fromJSON ''" <> BL.toStrict (A.encode attrs) <> "'' // { outPath = " <> argPath <> "; }"--  topDerivationPaths <- liftIO $ newIORef mempty--  emit <- makeEventEmitter writeToBatch--  let allowedPaths = toList inputLocations <&> toS <&> encodeUtf8-  adHocSystem <--    readFileMaybe (projectDir </> "ci-default-system.txt")-  liftIO (findNixFile projectDir) >>= \case-    Left e ->-      emit $-        EvaluateEvent.Message-          Message.Message-            { Message.index = -1, -- will be set by emit-              Message.typ = Message.Error,-              Message.message = e-            }-    Right file -> TraversalQueue.with $ \(derivationQueue :: Queue StorePath) ->-      let doIt =-            ( do-                Async.Lifted.concurrently_ evaluation emitDrvs-                -- derivationInfo upload has finished-                -- allAttrPaths :: IORef has been populated-            )-              `finally` pushDrvs-          uploadDrvInfos drvPath = do-            TraversalQueue.enqueue derivationQueue drvPath-            TraversalQueue.waitUntilDone derivationQueue-          addTopDerivation drvPath = do-            TraversalQueue.enqueue derivationQueue drvPath-            liftIO $ atomicModifyIORef topDerivationPaths ((,()) . S.insert drvPath)-          evaluation = do-            Nix.withExtraOptions [("system", T.strip s) | Just s <- [adHocSystem]] do-              let evalProc =-                    do runEvalProcess-                      store-                      projectDir-                      file-                      autoArguments-                      nixPath-                      captureAttrDrvAndEmit-                      uploadDrvInfos-                      sync-                      task-                      allowedPaths-              evalProc `finally` do-                -- Always upload drv infos, even in case of a crash in the worker-                TraversalQueue.waitUntilDone derivationQueue-                  `finally` TraversalQueue.close derivationQueue-          pushDrvs = do-            caches <- activePushCaches-            paths <- liftIO $ readIORef topDerivationPaths-            forM_ caches $ \cache -> do-              withNamedContext "cache" cache $ logLocM DebugS "Pushing derivations"-              Agent.Cache.push store cache (toList paths) pushEvalWorkers-              emit $ EvaluateEvent.PushedAll $ PushedAll.PushedAll {cache = cache}-          captureAttrDrvAndEmit msg = do-            case msg of-              EvaluateEvent.Attribute ae -> do-                storePath <- liftIO $ parseStorePath store (encodeUtf8 $ AttributeEvent.derivationPath ae)-                addTopDerivation storePath-              EvaluateEvent.AttributeEffect ae -> do-                storePath <- liftIO $ parseStorePath store (encodeUtf8 $ AttributeEffectEvent.derivationPath ae)-                addTopDerivation storePath-              _ -> pass-            emit msg-          emitDrvs =-            TraversalQueue.work derivationQueue $ \recurse drvPath -> do-              drvInfo <- retrieveDerivationInfo store drvPath-              for_ (M.keys $ DerivationInfo.inputDerivations drvInfo) \inp -> do-                inputStorePath <- liftIO $ parseStorePath store (encodeUtf8 inp)-                recurse inputStorePath -- asynchronously-              emit $ EvaluateEvent.DerivationInfo drvInfo-       in doIt+     in evaluation  isValidName :: FilePath -> Bool isValidName "" = False@@ -399,6 +579,7 @@ checkNonEmptyText t = Just t  runEvalProcess ::+  (Vector LogEntry -> IO ()) ->   CNix.Store ->   FilePath ->   FilePath ->@@ -413,9 +594,8 @@   EvaluateTask.EvaluateTask ->   [ByteString] ->   App ()-runEvalProcess store projectDir file autoArguments nixPath emit uploadDerivationInfos flush task allowedPaths = do+runEvalProcess sendLogItems store projectDir file autoArguments nixPath emit uploadDerivationInfos flush task allowedPaths = do   extraOpts <- Nix.askExtraOptions-  bulkBaseURL <- asks (ServiceInfo.bulkSocketBaseURL . Env.serviceInfo)   apiBaseUrl <- asks (toS . showBaseUrl . Env.herculesBaseUrl)   cfg <- asks Env.config   srcInput <- getSrcInput task@@ -466,12 +646,6 @@             Eval.file = toS file,             Eval.autoArguments = autoArguments,             Eval.extraNixOptions = extraOpts,-            Eval.logSettings =-              LogSettings.LogSettings-                { token = Sensitive $ EvaluateTask.logToken task,-                  path = "/api/v1/logs/build/socket",-                  baseURL = toS $ Network.URI.uriToString identity bulkBaseURL ""-                },             Eval.gitSource = ViaJSON gitSource,             Eval.srcInput = ViaJSON <$> srcInput,             Eval.apiBaseUrl = apiBaseUrl,@@ -518,7 +692,7 @@                   ("protocol.file.allow", "always")                 ]           }-  withProducer (produceWorkerEvents (EvaluateTask.id task) eval envSettings commandChan) $+  withProducer (produceWorkerEvents sendLogItems (EvaluateTask.id task) eval envSettings commandChan) $     \workerEventsP -> fix $ \continue ->       joinSTM $         listen@@ -622,7 +796,8 @@                         caches <- activePushCaches                         forM_ caches $ \cache -> do                           withNamedContext "cache" cache $ logLocM DebugS "Pushing derivations for import from derivation"-                          Agent.Cache.push store cache [storePath] pushEvalWorkers+                          withStorePush cache do+                            Agent.Cache.push store cache [storePath] pushEvalWorkers                   Async.Lifted.concurrently_                     (uploadDerivationInfos storePath)                     pushDerivations@@ -644,6 +819,9 @@                       uio <- askUnliftIO                       liftIO $ forkIO $ unliftIO uio doPoll                 continue+              Event.LogItems (ViaJSON e) -> do+                liftIO (sendLogItems e)+                continue               Event.OnPushHandler (ViaJSON e) -> do                 emit $ EvaluateEvent.OnPushHandlerEvent e                 continue@@ -682,17 +860,16 @@   pure $ toS $ Network.URI.uriRegName a  produceWorkerEvents ::-  Id (Task a) ->+  (Vector LogEntry -> IO ()) ->+  Id (Task EvaluateTask.EvaluateTask) ->   Eval.Eval ->   WorkerProcess.WorkerEnvSettings ->   Chan (Maybe Command.Command) ->   (Event.Event -> App ()) ->   App ExitCode-produceWorkerEvents taskId eval envSettings commandChan writeEvent = do+produceWorkerEvents sendLogEntries taskId eval envSettings commandChan writeEvent = do   workerExe <- WorkerProcess.getWorkerExe-  let opts = [show $ Eval.extraNixOptions eval]-  -- NiceToHave: replace renderNixPath by something structured like -I-  -- to support = and : in paths+  let opts = ["eval", fromMaybe "" eval.gitSource.fromViaJSON.webUrl, eval.gitSource.fromViaJSON.rev] <&> T.unpack   workerEnv <- liftIO $ WorkerProcess.prepareEnv envSettings   let wps =         (System.Process.proc workerExe opts)@@ -702,13 +879,15 @@           }       stderrHandler =         stderrLineHandler+          sendLogEntries           ( M.fromList               [ ("taskId", A.toJSON taskId),-                ("evalRev", A.toJSON (eval & Eval.gitSource & fromViaJSON & GitSource.rev))+                ("evalRev", A.toJSON (eval.gitSource.fromViaJSON.rev))               ]           )           "Effect worker"-  WorkerProcess.runWorker wps stderrHandler commandChan writeEvent+  cfg <- WorkerConfig.getWorkerConfig+  WorkerProcess.runWorker cfg wps stderrHandler commandChan writeEvent  drvPoller :: Maybe UUID -> Text -> App (UUID, BuildResult.BuildStatus) drvPoller notAttempt drvPath = do

@@ -8,6 +8,7 @@ import Hercules.Agent.Config.BinaryCaches qualified as BC import Hercules.Agent.Env (Env (Env)) import Hercules.Agent.Env qualified as Env+import Hercules.Agent.Memo (newMemo) import Hercules.Agent.Netrc.Env qualified as Netrc import Hercules.Agent.Nix.Init qualified import Hercules.Agent.SecureDirectory qualified as SecureDirectory@@ -36,6 +37,8 @@   let clientEnv :: Servant.Client.ClientEnv       clientEnv = Servant.Client.mkClientEnv manager baseUrl   token <- Token.readTokenFile $ toS $ Config.clusterJoinTokenPath config+  concPushes <- newMemo+  concQueries <- newMemo   withLogging $ Hercules.Agent.Cachix.Init.withEnv config (BC.cachixCaches bcs) \cachix -> liftIO do     nix <- Hercules.Agent.Nix.Init.newEnv     serviceInfo <- ServiceInfo.newEnv clientEnv@@ -54,7 +57,9 @@               kContext = mempty,               kLogEnv = logEnv,               nixEnv = nix,-              netrcEnv = Netrc.Env Nothing+              netrcEnv = Netrc.Env Nothing,+              concurrentStorePushes = concPushes,+              concurrentStoreQueries = concQueries             }     f env 

@@ -0,0 +1,23 @@+module Hercules.Agent.InitWorkerConfig where++import Data.Coerce (coerce)+import Hercules.Agent.Config qualified as Config+import Hercules.Agent.Env (App)+import Hercules.Agent.Env qualified+import Hercules.Agent.Nix qualified as Nix+import Hercules.Agent.WorkerProtocol.WorkerConfig (WorkerConfig (WorkerConfig))+import Hercules.Agent.WorkerProtocol.WorkerConfig qualified+import Hercules.CNix.Verbosity qualified as CNix.Verbosity+import Protolude++getWorkerConfig :: App WorkerConfig+getWorkerConfig = do+  verbosity_ <- asks (\env -> env.config.logLevel)+  nixVerbosity_ <- liftIO CNix.Verbosity.getVerbosity+  nixOptions_ <- Nix.askExtraOptions+  pure+    WorkerConfig+      { verbosity = coerce verbosity_,+        nixVerbosity = coerce nixVerbosity_,+        nixOptions = nixOptions_+      }

@@ -14,6 +14,9 @@ import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as LBS import Data.Map qualified as M+import Data.Vector (Vector)+import Hercules.API.Logs.LogEntry (LogEntry)+import Hercules.API.Logs.LogEntry qualified as LogEntry import Katip hiding (logLocM) import Katip.Core import Katip.Monadic (logLocM)@@ -31,16 +34,16 @@   logLocM ErrorS $ logStr msg   panic msg -stderrLineHandler :: (KatipContext m) => Map Text Value -> Text -> Int -> ByteString -> m ()-stderrLineHandler callerContext _processRole _ ln+makeStderrLogItem :: ByteString -> LogEntry+makeStderrLogItem msg = LogEntry.Msg {msg = decodeUtf8With lenientDecode msg, i = 0, ms = 0, level = 0}++stderrLineHandler :: (KatipContext m) => (Vector LogEntry -> IO ()) -> Map Text Value -> Text -> Int -> ByteString -> m ()+stderrLineHandler _sendLogItems 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) . extendContext <$> item)   where     extendContext workerItem = M.union workerItem callerContext-stderrLineHandler _ processRole pid ln =-  withNamedContext "worker" (pid :: Int) $-    logLocM InfoS $-      logStr $-        processRole <> ": " <> decodeUtf8With lenientDecode ln+stderrLineHandler sendLogItems _ _processRole _pid ln =+  liftIO $ sendLogItems $ pure $ makeStderrLogItem ln

@@ -0,0 +1,298 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE TypeFamilies #-}++module Hercules.Agent.LogSocket where++import Conduit+  ( ConduitM,+    ConduitT,+    Flush (..),+    await,+    awaitForever,+    filterC,+    fuseUpstream,+    mapC,+    runConduit,+    yield,+    (.|),+  )+import Data.Time (UTCTime, diffUTCTime, getCurrentTime)+import Data.Vector (Vector)+import Data.Vector qualified as V+import GHC.Conc (labelThread)+import Hercules.API.Agent.LifeCycle.ServiceInfo qualified+import Hercules.API.Logs.LogEntry (LogEntry ())+import Hercules.API.Logs.LogEntry qualified as LogEntry+import Hercules.API.Logs.LogHello (LogHello (LogHello))+import Hercules.API.Logs.LogHello qualified+import Hercules.API.Logs.LogMessage (LogMessage)+import Hercules.API.Logs.LogMessage qualified as LogMessage+import Hercules.Agent.Conduit (takeCWhileStopEarly, withMessageLimit)+import Hercules.Agent.Log (KatipContext, Severity (DebugS, ErrorS), katipAddContext, logLocM, sl)+import Hercules.Agent.Socket qualified as Socket+import Hercules.CNix.Store (getClientProtocolVersion)+import Katip (KatipContext (getKatipContext, getKatipNamespace), KatipContextT, Severity (WarningS), getLogEnv, runKatipContextT)+import Network.URI qualified+import Protolude hiding (atomically, finally, forkFinally, myThreadId, withAsync, yield)+import System.Timeout.Lifted (timeout)+import UnliftIO (MonadUnliftIO, atomically, catch, finally, lengthTBQueue, newTBQueueIO, readTBQueue, withAsync, writeTBQueue)+import UnliftIO.Concurrent (forkFinally, myThreadId)++data LogSettings = LogSettings+  { path :: !Text,+    baseURL :: !Text,+    token :: !Text+  }+  deriving (Generic)++withLoggerNoFlush ::+  (MonadUnliftIO m, KatipContext m) =>+  -- | Label for the logger thread+  Text ->+  -- | Store protocol version+  Int ->+  LogSettings ->+  -- | The logger function. It immediately replaces the time field of the log entries and queues them for asynchronous delivery.+  ((Vector LogEntry -> IO ()) -> m a) ->+  m a+withLoggerNoFlush label storeProtocolVersion settings f = do+  start <- liftIO getCurrentTime+  (loggr, stop) <- forkLogger label storeProtocolVersion settings+  f (loggr <=< setLogEntriesTime start) `finally` liftIO stop++setLogEntriesTime :: (Functor l) => UTCTime -> l LogEntry -> IO (l LogEntry)+setLogEntriesTime start l = do+  now <- getCurrentTime+  let diffMs = floor $ 1000 * diffUTCTime now start+  pure $ l <&> \e -> e {LogEntry.ms = diffMs}++forkLogger ::+  forall m.+  (MonadUnliftIO m, KatipContext m) =>+  -- | Label for the logger thread+  Text ->+  -- | Store protocol version+  Int ->+  LogSettings ->+  m (Vector LogEntry -> IO (), IO ())+forkLogger label storeProtocolVersion settings = katipAddContext (sl "label" label) do+  q <- newTBQueueIO 100000++  let entriesSource :: ConduitM () (Vector LogEntry) m ()+      entriesSource = do+        item <- lift do+          ( atomically do+              entries <- readTBQueue q+              pure entries+            )+            `UnliftIO.catch` ( \(e :: SomeException) -> do+                                 katipAddContext (sl "exception" (show e :: Text)) do+                                   logLocM ErrorS $ "exception in entriesSource"+                                   throwIO e+                             )+        case item of+          Nothing -> pass -- terminate conduit+          Just entries -> yield entries >> entriesSource++  _threadId <- flip+    forkFinally+    ( \case+        Right () -> logLocM DebugS "Logger done"+        Left e -> katipAddContext (sl "exception" (show e :: Text)) do+          logLocM ErrorS "Logger failed"+    )+    do+      t <- myThreadId+      liftIO (labelThread t ("logger for " <> toS label))+      logger settings storeProtocolVersion entriesSource+      pass+  (UnliftKatipContextT unliftKatip) <- askUnliftKatipContextT+  let putQ item = do+        l <- atomically do+          writeTBQueue q item+          lengthTBQueue q+        when (l > 10000) do+          unliftKatip $ logLocM WarningS "Logger queue has significant backlog."+  pure+    ( putQ . Just,+      putQ Nothing+    )++logger :: (MonadUnliftIO m, KatipContext m) => LogSettings -> Int -> ConduitM () (Vector LogEntry) m () -> m ()+logger logSettings_ storeProtocolVersionValue entriesSource = do+  socketConfig <- liftIO $ makeSocketConfig logSettings_ storeProtocolVersionValue+  let withPings socket m =+        withAsync+          ( liftIO $ forever do+              -- TODO add ping constructor to Frame or use websocket pings+              let ping = LogMessage.LogEntries mempty+              threadDelay 30_000_000+              atomically $ Socket.write socket ping+          )+          (const m)+  Socket.withReliableSocket socketConfig $ \socket -> withPings socket do+    let conduit =+          entriesSource+            .| unbatch+            .| filterProgress+            .| dropMiddle+            .| renumber 0+            .| batchAndEnd+            .| socketSink socket+        batch' = batch .| mapC (LogMessage.LogEntries . V.fromList)+        batchAndEnd =+          (foldMapTap (Last . ims) `fuseUpstream` batch') >>= \case+            Last (Just (i, ms)) -> yield $ LogMessage.End {i = i + 1, ms = ms}+            Last Nothing -> yield $ LogMessage.End 0 0+          where+            ims (Chunk logEntry) = Just (logEntry.i, logEntry.ms)+            ims _ = Nothing+        renumber i =+          await >>= traverse_ \case+            Flush -> yield Flush >> renumber i+            Chunk e -> do+              yield $ Chunk e {LogEntry.i = i}+              renumber (i + 1)+    runConduit conduit+    logLocM DebugS "Syncing"+    liftIO (timeout 600_000_000 $ Socket.syncIO socket) >>= \case+      Just _ -> pass+      Nothing -> panic "Could not push logs within 10 minutes after completion"+    logLocM DebugS "Logger done"++makeSocketConfig :: (MonadIO m) => LogSettings -> Int -> IO (Socket.SocketConfig LogMessage Hercules.API.Agent.LifeCycle.ServiceInfo.ServiceInfo m)+makeSocketConfig l storeProtocolVersionValue = do+  clientProtocolVersionValue <- liftIO getClientProtocolVersion+  baseURL_ <- case Network.URI.parseURI $ toS l.baseURL of+    Just x -> pure x+    Nothing -> panic "LogSettings: invalid base url"+  pure+    Socket.SocketConfig+      { makeHello =+          pure $+            LogMessage.Hello+              LogHello+                { clientProtocolVersion = clientProtocolVersionValue,+                  storeProtocolVersion = storeProtocolVersionValue+                },+        checkVersion = Socket.checkVersion',+        baseURL = baseURL_,+        path = l.path,+        token = encodeUtf8 l.token+      }++batch :: (Monad m) => ConduitT (Flush a) [a] m ()+batch = go []+  where+    go acc =+      await >>= \case+        Nothing -> do+          unless (null acc) (yield $ reverse acc)+        Just Flush -> do+          unless (null acc) (yield $ reverse acc)+          go []+        Just (Chunk c) -> do+          go (c : acc)++unbatch :: (Monad m, Foldable l) => ConduitT (l a) (Flush a) m ()+unbatch = awaitForever $ \l -> do+  for_ l $ \a -> yield $ Chunk a+  yield Flush++-- Sum must be < 100_000+richLogLimit, textOnlyLogLimit, tailLimit :: Int+richLogLimit = 40_000+textOnlyLogLimit = 49_900+tailLimit = 10_000++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++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++-- | Perform a foldMap while yielding the original values ("tap").+--+-- '<>' is invoked with the new b on the right.+foldMapTap :: (Monoid b, Monad m) => (a -> b) -> ConduitT a a m b+foldMapTap f = go mempty+  where+    go b =+      await >>= \case+        Nothing -> pure b+        Just a -> do+          yield a+          go (b <> f a)++-- TODO: Use 'nubProgress' instead?++-- | Remove spammy progress results.+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++----- Utilities -----++-- | Captures the whole logging context, by providing the capability to reduce @KatipContextT m@ to a plain @m@.+newtype UnliftKatipContextT = UnliftKatipContextT {unliftKatipContextT :: forall a m. KatipContextT m a -> m a}++askUnliftKatipContextT :: (KatipContext m) => m UnliftKatipContextT+askUnliftKatipContextT = do+  le <- getLogEnv+  ctx <- getKatipContext+  ns <- getKatipNamespace+  pure (UnliftKatipContextT (runKatipContextT le ctx ns))

@@ -0,0 +1,124 @@+{-# LANGUAGE BlockArguments #-}++module Hercules.Agent.Memo+  ( Memo,+    newMemo,+    query,+    doOnce,+    multiQuery,+  )+where++import Data.Map qualified as M+import Protolude hiding (atomically, bracket, throwIO)+import UnliftIO+  ( MonadUnliftIO,+    TVar,+    atomically,+    bracket,+    modifyTVar,+    newEmptyTMVarIO,+    newTVarIO,+    putTMVar,+    readTMVar,+    readTVar,+    throwIO,+    writeTVar,+  )++data Entry v+  = Result v+  | Promise (STM (Either SomeException v))++-- | An unbounded cache. No coherent function/action - passed by callers instead. Use with care.+--+-- * 'MonadUnliftIO'+-- * process multiple keys at once+-- * anti dogpiling+-- * input not restricted to key, for efficiency and practicality+newtype Memo k v = Memo (TVar (Map k (Entry v)))++newMemo :: (MonadIO m) => m (Memo k v)+newMemo = liftIO do+  Memo <$> newTVarIO M.empty++-- | Like 'query' but you pass the key through the lexical scope.+doOnce :: (Show k, Ord k, MonadUnliftIO m) => Memo k v -> k -> m v -> m v+doOnce memo key action = query memo (const action) key++query :: (Show k, Ord k, MonadUnliftIO m) => Memo k v -> (k -> m v) -> k -> m v+query memo handler k = do+  query'+    memo+    ( \m -> do+        unless (m == M.singleton k ()) do+          panic $ "query: key " <> show k <> " must be the only request"+        M.singleton k <$> handler k+    )+    k++query' :: (Show k, Ord k, MonadUnliftIO m) => Memo k v -> (Map k () -> m (Map k v)) -> k -> m v+query' memo handler k = do+  x <- multiQuery memo handler (M.singleton k ())+  case M.lookup k x of+    Nothing -> panic $ "query: key " <> show k <> " must occur in response"+    Just v -> pure v++-- | Perform a query such that no item is queried more than once.+multiQuery ::+  (Show k, Ord k, MonadUnliftIO m) =>+  Memo k v ->+  -- | Function to perform the query. This should always behave the same.+  (Map k () -> m (Map k v)) ->+  -- | The items to query+  Map k () ->+  -- | Deduplicated query action.+  m (Map k v)+multiQuery (Memo cacheVar) performQuery inputs = do+  -- Overview+  -- 1. Add promises to the map+  -- 2. Run the query for the new promises+  -- 3. Replace the promises by values so they're not removed by the bracket cleanup+  -- 4. Query the cache which now has the required entries+  multiQueryPromise <- newEmptyTMVarIO+  let makePromises = do+        M.mapWithKey \k _ -> Promise do+          mp <- readTMVar multiQueryPromise+          pure case M.lookup k mp of+            Nothing -> Left $ toException $ FatalError $ "multiQuery: input " <> show k <> " does not occur in query result"+            Just v -> Right v+      putPromises = do+        cache <- readTVar cacheVar+        let newInputs = M.difference inputs cache+            newPromises = makePromises newInputs+        writeTVar cacheVar (cache `M.union` newPromises)+        pure newPromises+      putResults rs = do+        cache <- readTVar cacheVar+        writeTVar cacheVar (fmap Result rs `M.union` cache)+      queryCache = do+        cache <- atomically $ readTVar cacheVar+        inputs & M.traverseWithKey \k _unit -> do+          case M.lookup k cache of+            Just (Result x) -> pure x+            Just (Promise x) -> do+              v' <- atomically x+              either throwIO pure v'+            Nothing -> panic $ "multiQuery: input " <> show k <> " was not saved to cache"+      clearPromises = \promises -> do+        modifyTVar cacheVar \c ->+          let cacheContainsPromise k _v = case M.lookup k c of+                Just (Promise _) -> True+                _ -> False+           in M.difference c (M.filterWithKey cacheContainsPromise promises)++  bracket+    (atomically putPromises)+    (atomically . clearPromises)+    \promises -> do+      when (not (null promises)) do+        response <- performQuery (void promises)+        atomically do+          putTMVar multiQueryPromise response+          putResults response+      queryCache

@@ -31,12 +31,12 @@  parseConfigPath :: Parser ConfigPath parseConfigPath =-  TomlPath+  ConfigPath     <$> strOption       ( long "config"           <> metavar "FILE"           <> help-            "File path to the configuration file (TOML)"+            "File path to the configuration file (TOML, or JSON if the path ends in .json)"       )  parserInfo :: ParserInfo Options

@@ -0,0 +1,23 @@+-- | Limit concurrent access to a resource. Wrapper around 'TSem`.+module Hercules.Agent.ResourceLimiter where++import Control.Concurrent.STM.TSem (TSem, newTSem, signalTSem, waitTSem)+import Protolude hiding (atomically, bracket_)+import UnliftIO (MonadUnliftIO, atomically, bracket_)++-- | Limit concurrent access to a resource. Wrapper around 'TSem` semaphore.+newtype ResourceLimiter = ResourceLimiter TSem++-- | Create a semaphore.+newResourceLimiter :: (MonadIO m) => Integer -> m ResourceLimiter+newResourceLimiter n =+  if n < 1+    then panic "newResourceLimiter: concurrency must be at least 1"+    else atomically $ do+      ResourceLimiter <$> newTSem n++-- | Perform an action while holding a resource.+--+-- To avoid priority inversion and deadlock, use this around small operations, preferably.+withResource :: (MonadUnliftIO m) => ResourceLimiter -> m a -> m a+withResource (ResourceLimiter sem) = bracket_ (atomically (waitTSem sem)) (atomically (signalTSem sem))

@@ -15,6 +15,10 @@  using namespace nix; +#if NIX_IS_AT_LEAST(2,19,0)+#include <nix/signals.hh>+#endif+ #if ! NIX_IS_AT_LEAST(2,15,0) using nix::daemon::TrustedFlag; using nix::daemon::NotTrusted;

@@ -1,35 +1,11 @@ module Data.Conduit.Extras where  import Control.Concurrent.Chan-import Control.Exception import Control.Monad.IO.Class-import Control.Monad.IO.Unlift import Control.Monad.Trans import Data.Conduit import Prelude --- | Write to a channel terminating with @Nothing@-sinkChanTerminate :: (MonadUnliftIO m, MonadIO m) => Chan (Maybe a) -> ConduitT a o m ()-sinkChanTerminate ch =-  handleC-    ( \e -> liftIO $ do-        writeChan ch Nothing-        throwIO (e :: SomeException)-    )-    $ do-      awaitForever $ \msg -> liftIO $ writeChan ch (Just msg)-      liftIO $ writeChan ch Nothing---- | Write to a channel terminating with @Nothing@-sinkChan :: (MonadUnliftIO m, MonadIO m) => Chan (Maybe a) -> ConduitT a o m ()-sinkChan ch =-  handleC-    ( \e -> liftIO $ do-        throwIO (e :: SomeException)-    )-    $ do-      awaitForever $ \msg -> liftIO $ writeChan ch (Just msg)- -- | Read a channel until @Nothing@ is encountered sourceChan :: (MonadIO m) => Chan (Maybe a) -> ConduitT i a m () sourceChan ch = do@@ -38,6 +14,6 @@     Nothing -> liftIO $ writeChan ch Nothing     Just msg -> yield msg >> sourceChan ch -conduitToCallbacks :: (MonadUnliftIO m, MonadIO m) => ConduitT () o m a -> (o -> m ()) -> m a+conduitToCallbacks :: (MonadIO m) => ConduitT () o m a -> (o -> m ()) -> m a conduitToCallbacks c w = do   runConduit (c `fuseUpstream` awaitForever (lift . w))

@@ -0,0 +1,78 @@+{-# LANGUAGE BlockArguments #-}++module Hercules.Agent.Conduit where++import Data.Conduit (ConduitT, await, awaitForever, yield, (.|))+import Data.IORef (IORef, modifyIORef, newIORef, readIORef)+import Data.Sequence qualified 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

@@ -33,6 +33,7 @@  import Control.Monad.Trans.Maybe (MaybeT (MaybeT), runMaybeT) import Data.Map qualified as M+import Data.Type.Equality (type (~)) import Hercules.API.Agent.Evaluate.EvaluateEvent.InputDeclaration (InputDeclaration (SiblingInput), SiblingInput (MkSiblingInput)) import Hercules.API.Agent.Evaluate.EvaluateEvent.InputDeclaration qualified import Hercules.Agent.NixFile.CiNixArgs (CiNixArgs (CiNixArgs))

@@ -1,21 +1,11 @@ {-# LANGUAGE DerivingVia #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# OPTIONS_GHC -Wno-orphans #-} -module Hercules.Agent.Sensitive where+module Hercules.Agent.Sensitive (module Hercules.API.Sensitive) where  import Data.Binary-import Protolude-import Text.Show---- | newtype wrapper to avoid leaking sensitive data through Show-newtype Sensitive a = Sensitive {reveal :: a}-  deriving (Generic)-  deriving newtype (Binary, Eq, Ord, Monoid, Semigroup)-  deriving (Functor, Applicative, Monad) via Identity---- | @const "<sensitive>"@-instance Show (Sensitive a) where-  show _ = "<sensitive>"+import Hercules.API.Sensitive -revealContainer :: (Functor f) => Sensitive (f a) -> f (Sensitive a)-revealContainer (Sensitive fa) = Sensitive <$> fa+deriving newtype instance (Binary a) => Binary (Sensitive a)

@@ -64,7 +64,7 @@ ackTimeout :: NominalDiffTime ackTimeout = 60 -- seconds -withReliableSocket :: (A.FromJSON sp, A.ToJSON ap, MonadIO m, MonadUnliftIO m, KatipContext m) => SocketConfig ap sp m -> (Socket sp ap -> m a) -> m a+withReliableSocket :: (A.FromJSON sp, A.ToJSON ap, MonadUnliftIO m, KatipContext m) => SocketConfig ap sp m -> (Socket sp ap -> m a) -> m a withReliableSocket socketConfig f = do   writeQueue <- atomically $ newTBQueue 100   agentMessageNextN <- newTVarIO 0@@ -281,7 +281,7 @@ slash :: [Char] -> [Char] -> [Char] a `slash` b = dropWhileEnd (== '/') a <> "/" <> dropWhile (== '/') b -withTimeout :: (Exception e, MonadIO m, MonadUnliftIO m) => NominalDiffTime -> e -> m a -> m a+withTimeout :: (Exception e, MonadUnliftIO m) => NominalDiffTime -> e -> m a -> m a withTimeout t e _ | t <= 0 = throwIO e withTimeout t e m =   timeout (ceiling $ t * 1_000_000) m >>= \case

@@ -0,0 +1,24 @@+module Hercules.Agent.Store where++import Hercules.Agent.WorkerProtocol.OutputInfo (OutputInfo (OutputInfo))+import Hercules.Agent.WorkerProtocol.OutputInfo qualified+import Hercules.CNix.Store (DerivationOutput (derivationOutputName, derivationOutputPath), Store, getStorePathBaseName, queryPathInfo, validPathInfoNarHash32, validPathInfoNarSize, validPathInfoReferences')+import Hercules.CNix.Store qualified as CNix+import Protolude++toDrvInfo :: Store -> DerivationOutput -> IO OutputInfo+toDrvInfo store drvOut = do+  -- FIXME: ca-derivations: always get the built path+  vpi <- for (derivationOutputPath drvOut) (queryPathInfo store)+  hash_ <- traverse validPathInfoNarHash32 vpi+  path <- traverse (CNix.storePathToPath store) (derivationOutputPath drvOut)+  let size = fmap validPathInfoNarSize vpi+  refs <- traverse (traverse getStorePathBaseName <=< validPathInfoReferences') vpi+  pure+    OutputInfo+      { name = derivationOutputName drvOut,+        path = fromMaybe "" path,+        hash = fromMaybe "" hash_,+        size = fromMaybe 0 size,+        references = fromMaybe [] refs+      }

@@ -11,6 +11,7 @@ import Conduit hiding (Producer) import Control.Exception.Safe qualified as Safe import Control.Monad.IO.Unlift+import Data.Aeson qualified as A import Data.Binary (Binary) import Data.Conduit.Extras (conduitToCallbacks, sourceChan) import Data.Conduit.Serialization.Binary@@ -21,13 +22,20 @@ import GHC.IO.Exception import Hercules.API.Agent.Evaluate.EvaluateTask qualified as EvaluateTask import Hercules.Agent.NixPath (renderNixPath)+import Hercules.Agent.WorkerProtocol.WorkerConfig (WorkerConfig) import Paths_hercules_ci_agent (getBinDir)-import Protolude+import Protolude hiding (yield) import System.Environment (getEnvironment) import System.FilePath ((</>)) import System.IO (hClose)-import System.IO.Error+import System.IO.Error (illegalOperationErrorType, ioeGetErrorType, mkIOError) import System.Process+  ( CreateProcess (std_err, std_in, std_out),+    StdStream (CreatePipe),+    getPid,+    waitForProcess,+    withCreateProcess,+  ) import System.Timeout (timeout) import Prelude () @@ -75,13 +83,15 @@ -- using a 'Binary' interface. runWorker ::   (Binary command, Binary event, MonadUnliftIO m, MonadThrow m) =>+  -- | Extra Nix options, and other "env" values+  WorkerConfig ->   -- | Process invocation details. Will ignore std_in, std_out and std_err fields.   CreateProcess ->   (Int -> ByteString -> m ()) ->   Chan (Maybe command) ->   (event -> m ()) ->   m ExitCode-runWorker baseProcess stderrLineHandler commandChan eventHandler = do+runWorker workerConfig baseProcess stderrLineHandler commandChan eventHandler = do   UnliftIO {unliftIO = unlift} <- askUnliftIO   let createProcessSpec =         baseProcess@@ -109,8 +119,11 @@                 (sourceHandle errHandle .| linesUnboundedAsciiC .| awaitForever (liftIO . unlift . stderrLineHandler pid))       let eventConduit = sourceHandle outHandle .| conduitDecode           commandConduit =-            sourceChan commandChan-              .| conduitEncode+            ( do+                yield (toS (A.encode workerConfig <> "\n"))+                sourceChan commandChan+                  .| conduitEncode+            )               .| concatMapC (\x -> [Chunk x, Flush])               .| handleC handleEPIPE (sinkHandleFlush inHandle)           handleEPIPE e | ioeGetErrorType e == ResourceVanished = pure ()

@@ -3,13 +3,12 @@ module Hercules.Agent.WorkerProtocol.Command.Build where  import Data.Binary-import Hercules.Agent.WorkerProtocol.LogSettings import Protolude  data Build = Build   { drvPath :: Text,     inputDerivationOutputPaths :: [ByteString],-    logSettings :: LogSettings,-    materializeDerivation :: Bool+    materializeDerivation :: Bool,+    materializePlatforms :: [ByteString]   }   deriving (Generic, Binary, Show, Eq)

@@ -7,23 +7,22 @@ import Data.Binary import Hercules.API.Id (Id) import Hercules.Agent.Sensitive-import Hercules.Agent.WorkerProtocol.LogSettings import Hercules.Agent.WorkerProtocol.Orphans () import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON)+import Hercules.Formats.Mountable (Mountable) import Hercules.Secrets (SecretContext) import Protolude  data Effect = Effect   { drvPath :: Text,     apiBaseURL :: Text,-    logSettings :: LogSettings,     inputDerivationOutputPaths :: [ByteString],-    materializeDerivation :: Bool,     secretsPath :: FilePath,     serverSecrets :: Sensitive (ViaJSON (Map Text (Map Text A.Value))),     token :: Sensitive Text,     projectId :: Id "project",     projectPath :: Text,-    secretContext :: SecretContext+    secretContext :: SecretContext,+    configuredMountables :: Sensitive (ViaJSON (Map Text (Mountable)))   }   deriving (Generic, Binary, Show, Eq)

@@ -6,7 +6,6 @@ import Hercules.API.Agent.Evaluate.EvaluateTask qualified as EvaluateTask import Hercules.API.Agent.Evaluate.ImmutableGitInput (ImmutableGitInput) import Hercules.Agent.NixFile.GitSource (GitSource)-import Hercules.Agent.WorkerProtocol.LogSettings import Hercules.Agent.WorkerProtocol.Orphans () import Hercules.Agent.WorkerProtocol.ViaJSON (ViaJSON) import Protolude@@ -15,14 +14,11 @@   { cwd :: FilePath,     file :: Text,     autoArguments :: Map Text Arg,-    -- | NB currently the options will leak from one evaluation to-    --   the next if you're running them in the same worker!-    --   (as of now, we use one worker process per evaluation)+    -- TODO: Also set at worker start, so remove this here to avoid ambiguity?     extraNixOptions :: [(Text, Text)],     gitSource :: ViaJSON GitSource,     srcInput :: Maybe (ViaJSON ImmutableGitInput),     apiBaseUrl :: Text,-    logSettings :: LogSettings,     selector :: ViaJSON EvaluateTask.Selector,     isFlakeJob :: Bool,     ciSystems :: Maybe (Map Text ()),

@@ -5,8 +5,10 @@  import Data.Binary (Binary) import Data.UUID (UUID)+import Data.Vector (Vector) import Hercules.API.Agent.Evaluate.EvaluateEvent.OnPushHandlerEvent (OnPushHandlerEvent) import Hercules.API.Agent.Evaluate.EvaluateEvent.OnScheduleHandlerEvent (OnScheduleHandlerEvent)+import Hercules.API.Logs.LogEntry (LogEntry) import Hercules.Agent.WorkerProtocol.Event.Attribute (Attribute) import Hercules.Agent.WorkerProtocol.Event.AttributeError (AttributeError) import Hercules.Agent.WorkerProtocol.Event.AttributeIFD (AttributeIFD)@@ -27,4 +29,5 @@   | OnPushHandler (ViaJSON OnPushHandlerEvent)   | OnScheduleHandler (ViaJSON OnScheduleHandlerEvent)   | Exception Text+  | LogItems (ViaJSON (Vector LogEntry))   deriving (Generic, Binary, Show, Eq)

@@ -3,21 +3,10 @@ module Hercules.Agent.WorkerProtocol.Event.BuildResult where  import Data.Binary+import Hercules.Agent.WorkerProtocol.OutputInfo (OutputInfo) import Protolude  data BuildResult   = BuildFailure {errorMessage :: Text}   | BuildSuccess {outputs :: [OutputInfo]}-  deriving (Generic, Binary, Show, Eq)--data OutputInfo = OutputInfo-  { -- | e.g. out, dev-    name :: ByteString,-    -- | store path-    path :: ByteString,-    -- | typically sha256:...-    hash :: ByteString,-    -- | nar size in bytes-    size :: Int64-  }   deriving (Generic, Binary, Show, Eq)

@@ -1,15 +0,0 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DerivingVia #-}--module Hercules.Agent.WorkerProtocol.LogSettings where--import Data.Binary-import Hercules.Agent.Sensitive-import Protolude--data LogSettings = LogSettings-  { path :: Text,-    baseURL :: Text,-    token :: Sensitive Text-  }-  deriving (Generic, Binary, Show, Eq)

@@ -0,0 +1,20 @@+{-# LANGUAGE DeriveAnyClass #-}++module Hercules.Agent.WorkerProtocol.OutputInfo where++import Data.Binary+import Protolude++data OutputInfo = OutputInfo+  { -- | e.g. out, dev+    name :: ByteString,+    -- | store path+    path :: ByteString,+    -- | typically sha256:...+    hash :: ByteString,+    -- | nar size in bytes+    size :: Int64,+    -- | references, in store path basename format+    references :: [ByteString]+  }+  deriving (Generic, Binary, Show, Eq)

@@ -0,0 +1,38 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE GeneralisedNewtypeDeriving #-}++module Hercules.Agent.WorkerProtocol.WorkerConfig where++import Control.Monad.Fail (fail)+import Data.Aeson (FromJSON (..), ToJSON (..))+import Hercules.CNix.Verbosity qualified+import Katip qualified+import Protolude++-- | Sets up the worker environment.+data WorkerConfig = WorkerConfig+  { -- | Verbosity+    verbosity :: !(ViaShowRead Katip.Severity),+    -- | Nix Verbosity+    nixVerbosity :: !(ViaShowRead Hercules.CNix.Verbosity.Verbosity),+    -- | Nix Options+    nixOptions :: ![(Text, Text)]+  }+  deriving (Generic, Show)+  -- This uses JSON so that we can getLine and parse it in the worker before+  -- doing any command parsing. Maybe we don't need framing (a line) and we+  -- can switch to Binary?+  deriving anyclass (ToJSON, FromJSON)++newtype ViaShowRead a = ViaShowRead {unViaShowRead :: a}+  deriving newtype (Generic, Show, Read, Eq, Ord)++instance (Show a) => ToJSON (ViaShowRead a) where+  toJSON = toJSON @[Char] . show . unViaShowRead++instance (Read a) => FromJSON (ViaShowRead a) where+  parseJSON v = do+    s <- parseJSON @[Char] v+    case readMaybe s of+      Nothing -> fail $ "Could not parse " <> show s+      Just a -> pure $ ViaShowRead a

@@ -4,12 +4,15 @@  module Hercules.Effect where +import Control.Concurrent.Async (AsyncCancelled (AsyncCancelled))+import Control.Exception.Safe (isAsyncException) import Control.Monad.Catch (MonadThrow) import Data.Aeson qualified as A import Data.Aeson.KeyMap qualified as AK import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BL import Data.Map qualified as M+import Data.Text qualified as T import Hercules.API.Agent.Evaluate.EvaluateEvent.AttributeEffectEvent (GitToken (..), SecretRef (GitToken, SimpleSecret), SimpleSecret (MkSimpleSecret)) import Hercules.API.Agent.Evaluate.EvaluateEvent.AttributeEffectEvent qualified import Hercules.API.Id (Id, idText)@@ -20,24 +23,39 @@ import Hercules.Effect.Container (BindMount (BindMount)) import Hercules.Effect.Container qualified as Container import Hercules.Error (escalateAs)+import Hercules.Formats.Mountable (Mountable)+import Hercules.Formats.Mountable qualified as Mountable import Hercules.Formats.Secret qualified as Formats.Secret import Hercules.Secrets (SecretContext, evalCondition, evalConditionTrace) import Katip (KatipContext, Severity (..), logLocM, logStr) import Network.Socket (Family (AF_UNIX), SockAddr (SockAddrUnix), SocketType (Stream), bind, listen, socket, withFdSocket) import Protolude import System.FilePath+import System.IO.Error (isDoesNotExistError) import System.Posix (dup, fdToHandle)+import System.Posix.Signals (killProcess, signalProcess)+import System.Posix.User (getEffectiveGroupID, getEffectiveUserID)+import System.Process (ProcessHandle)+import System.Process.Internals qualified as Process.Internal import UnliftIO.Directory (createDirectory, createDirectoryIfMissing) import UnliftIO.Process (withCreateProcess) import UnliftIO.Process qualified as Process +parseDrvMountsMap :: Map ByteString ByteString -> Either Text (Map Text Text)+parseDrvMountsMap drvEnv =+  case "__hci_effect_mounts" `M.lookup` drvEnv of+    Nothing -> pure mempty+    Just mountsMapText -> case A.eitherDecode (BL.fromStrict mountsMapText) of+      Left e -> Left $ "Could not parse __hci_effect_mounts variable in derivation. Error: " <> toS e+      Right r -> pure r+ parseDrvSecretsMap :: Map ByteString ByteString -> Either Text (Map Text SecretRef) parseDrvSecretsMap drvEnv =   case (,) "secretsToUse" <$> M.lookup "secretsToUse" drvEnv     <|> (,) "secretsMap" <$> M.lookup "secretsMap" drvEnv of     Nothing -> pure mempty     Just (attrName, secretsMapText) -> case A.eitherDecode (BL.fromStrict secretsMapText) of-      Left _ -> Left $ "Could not parse " <> attrName <> " variable in derivation. It must be a JSON dictionary."+      Left e -> Left $ "Could not parse " <> attrName <> " variable in derivation. Error: " <> toS e       Right r -> parseSecretRefs attrName r  parseSecretRefs :: Text -> A.Object -> Either Text (Map Text SecretRef)@@ -141,6 +159,7 @@     runEffectSecretsConfigPath :: Maybe FilePath,     runEffectSecretContext :: Maybe SecretContext,     runEffectServerSecrets :: Sensitive (Map Text (Map Text A.Value)),+    runEffectConfiguredMountables :: Sensitive (Map Text Mountable),     runEffectApiBaseURL :: Text,     runEffectDir :: FilePath,     runEffectProjectId :: Maybe (Id "project"),@@ -160,11 +179,12 @@   drvArgs <- liftIO $ getDerivationArguments derivation   drvEnv <- liftIO $ getDerivationEnv derivation   drvSecretsMap <- escalateAs FatalError $ parseDrvSecretsMap drvEnv+  drvMountsMap <- escalateAs FatalError $ parseDrvMountsMap drvEnv   let mkDir d = let newDir = dir </> d in toS newDir <$ createDirectory newDir   buildDir <- mkDir "build"   etcDir <- mkDir "etc"   secretsDir <- mkDir "secrets"-  runcDir <- mkDir "runc-state"+  containerDir <- mkDir "container-state"   let extraSecrets =         runEffectToken p           & maybe@@ -216,6 +236,20 @@               ("TMP", "/build"),               ("TEMP", "/build")             ]++        uid_ = drvEnv' & M.lookup "__hci_effect_virtual_uid" & maybe 0 (readOrThrow "__hci_effect_virtual_uid as integer")+        gid_ = drvEnv' & M.lookup "__hci_effect_virtual_gid" & maybe uid_ (readOrThrow "__hci_effect_virtual_gid as integer")+        rootReadOnly_ = drvEnv' & M.lookup "__hci_effect_root_read_only" & maybe False (readBool "__hci_effect_root_read_only")++        readOrThrow what str = case reads (toS str) of+          [(x, "")] -> x+          _ -> panic ("Could not parse " <> what <> " from derivation environment. Value was: " <> show str)+        readBool _ "1" = True+        readBool _ "" = False+        readBool _ "true" = True+        readBool _ "false" = False+        readBool what x = panic ("Could not parse boolean " <> what <> " from derivation environment. Value was: " <> show x)+         (//) :: (Ord k) => Map k a -> Map k a -> Map k a         (//) = flip M.union     let (withNixDaemonProxyPerhaps, forwardedSocketPath) =@@ -225,26 +259,82 @@                in (withNixDaemonProxy (runEffectExtraNixOptions p) socketPath, socketPath)             else (identity, "/nix/var/nix/daemon-socket/socket") +    -- Tell the runtime to use the current process' uid/gid+    hostUID_ <-+      getEffectiveUserID+        >>= \case+          0 -> panic "Refusing to host effect as root user"+          x -> pure x+    hostGID_ <-+      getEffectiveGroupID+        >>= \case+          x -> pure x++    extraBindMounts_ <- checkMounts (reveal $ runEffectConfiguredMountables p) (runEffectSecretContext p) drvMountsMap+    -- We've validated that the paths are pretty much canonical; otherwise this check would be insufficient.+    let isExtraBind path = extraBindMounts_ & any (\m -> Container.pathInContainer m == path)     withNixDaemonProxyPerhaps $       Container.run-        runcDir+        containerDir         Container.Config           { extraBindMounts =               [ BindMount {pathInContainer = "/build", pathInHost = buildDir, readOnly = False},-                BindMount {pathInContainer = "/etc", pathInHost = etcDir, readOnly = False},                 BindMount {pathInContainer = "/secrets", pathInHost = secretsDir, readOnly = True},-                -- we cannot bind mount this read-only because of https://github.com/opencontainers/runc/issues/1523-                BindMount {pathInContainer = "/etc/resolv.conf", pathInHost = "/etc/resolv.conf", readOnly = False},                 BindMount {pathInContainer = "/nix/var/nix/daemon-socket/socket", pathInHost = toS forwardedSocketPath, readOnly = True}-              ],+              ]+                ++ [ BindMount {pathInContainer = "/etc", pathInHost = etcDir, readOnly = False}+                     | not (isExtraBind "/etc")+                   ]+                ++ [+                     -- TODO: does this apply to crun?+                     -- we cannot bind mount this read-only because of https://github.com/opencontainers/runc/issues/1523+                     BindMount {pathInContainer = "/etc/resolv.conf", pathInHost = "/etc/resolv.conf", readOnly = False}+                     | not (isExtraBind "/etc") && not (isExtraBind "/etc/resolv.conf")+                   ]+                ++ extraBindMounts_,             executable = decodeUtf8With lenientDecode drvBuilder,             arguments = map (decodeUtf8With lenientDecode) drvArgs,             environment = overridableEnv // drvEnv' // onlyImpureOverridableEnv // impureEnvVars // fixedEnv,             workingDirectory = "/build",             hostname = "hercules-ci",-            rootReadOnly = False+            rootReadOnly = rootReadOnly_,+            virtualUID = uid_,+            virtualGID = gid_,+            hostUID = fromIntegral hostUID_,+            hostGID = fromIntegral hostGID_           } +checkMounts :: Map Text Mountable -> Maybe SecretContext -> Map Text Text -> IO [BindMount]+checkMounts configuredMnts secretContext drvMounts = do+  concat <$> forM (M.toList drvMounts) \(mntPath, mntName) -> do+    let -- Intentionally generic message, as misconfigurations on the agent are somewhat sensitive.+        abort = throwIO $ FatalError $ "While configuring the mount for effect sandbox path " <> show mntPath <> ", a mountable with name " <> mntName <> " has not been configured on agent, or it has been configured, but the condition field does not allow it to be used by this effect invocation. Make sure that mountable " <> mntName <> " exists in the agent configuration and that its condition field allows it to be used in the context of this job."+    case M.lookup mntName configuredMnts of+      Nothing -> do+        abort+      Just mountable -> do+        when (not ("/" `T.isPrefixOf` mntPath)) do+          throwIO $ FatalError ("Mount path must be absolute, but path does not start with /: " <> show mntPath)+        when ("/." `T.isInfixOf` mntPath) do+          throwIO $ FatalError ("Mount path must not contain /., but path is: " <> show mntPath)+        when ("//" `T.isInfixOf` mntPath) do+          throwIO $ FatalError ("Mount path must not contain //, but path is: " <> show mntPath)+        let -- Only valid after checks above+            checkPrefix path = do+              when (mntPath == path || (path <> "/") `T.isPrefixOf` mntPath) do+                throwIO $ FatalError ("Mount over " <> path <> " is not allowed: " <> show mntPath)+        checkPrefix "/nix"+        checkPrefix "/secrets"+        checkPrefix "/build"++        let cond = Mountable.condition mountable+        -- TODO: make hci effect run allow this? allow this when condition is simply `true`?+        ctx <- maybe (panic "No job context provided - don't know whether mounts are allowed.") pure secretContext+        let conditionOk = evalCondition ctx cond+        when (not conditionOk) do+          abort+        pure [BindMount {pathInContainer = mntPath, pathInHost = Mountable.source mountable, readOnly = Mountable.readOnly mountable}]+ withNixDaemonProxy :: [(Text, Text)] -> FilePath -> IO a -> IO a withNixDaemonProxy extraNixOptions socketPath wrappedAction = do   -- Open the socket asap, so we don't have to wait for@@ -269,5 +359,43 @@             Process.std_err = Process.Inherit,             Process.std_out = Process.UseHandle stderr           }-  withCreateProcess procSpec $ \_in _out _err _processHandle -> do+  withCreateProcess procSpec $ \_in _out _err processHandle -> do     wrappedAction+      -- TODO kill process _group_?+      `finally` destroyProcess_1s processHandle++forPid :: (Num pid) => ProcessHandle -> (pid -> IO ()) -> IO ()+forPid ph f = Process.Internal.withProcessHandle ph \case+  Process.Internal.OpenHandle {phdlProcessHandle = pid} -> f (fromIntegral pid)+  Process.Internal.OpenExtHandle {phdlProcessHandle = pid} -> f (fromIntegral pid)+  Process.Internal.ClosedHandle {} -> pure ()++-- | Make sure a process is terminated, in about 1s or less.+destroyProcess_1s :: ProcessHandle -> IO ()+destroyProcess_1s ph = do+  Process.terminateProcess ph+  let waitp = Process.waitForProcess ph+      killp = do+        threadDelay 500_000+        Process.terminateProcess ph+        threadDelay 500_000+        forPid ph \pid ->+          signalProcess killProcess pid+      handleKillException =+        handleJust+          ( \e -> case fromException e of+              -- completely ignore when cancelled by waitp completing early+              Just AsyncCancelled -> Nothing+              -- completely ignore asynchronous exceptions+              Nothing | isAsyncException e -> Nothing+              -- process may not exist anymore when sending a second signal (or more)+              -- in which case usually waitp will catch that during our threadDelay,+              -- but we shouldn't rely on that, as we don't want to raise false positives.+              Nothing | Just e' <- fromException e, isDoesNotExistError e' -> Nothing+              Nothing -> Just e+          )+          ( \e -> do+              -- TODO katip+              putErrLn ("hercules-ci-agent: Ignoring exception while stopping nix-daemon proxy " <> displayException e)+          )+  race_ waitp (killp & handleKillException)

@@ -11,6 +11,7 @@ import Data.ByteString.Lazy qualified as BL import Data.Map qualified as M import Data.UUID.V4 qualified as UUID+import Data.Vector qualified as V import GHC.IO.Exception (IOErrorType (HardwareFault)) import Protolude import System.Directory (createDirectory)@@ -38,11 +39,15 @@     environment :: Map Text Text,     workingDirectory :: Text,     hostname :: Text,-    rootReadOnly :: Bool+    rootReadOnly :: Bool,+    virtualUID :: Int,+    virtualGID :: Int,+    hostUID :: Int,+    hostGID :: Int   } -effectToRuncSpec :: Config -> Value -> Value-effectToRuncSpec config spec =+effectToOCIRuntimeSpec :: Config -> Value -> Value+effectToOCIRuntimeSpec config spec =   let defaultMounts = [defaultBindMount "/nix/store"]       mounts =         foldMap@@ -66,36 +71,55 @@         & key "process" . key "terminal" .~ toJSON True         & key "process" . key "env" .~ toJSON (config & environment & M.toList <&> \(k, v) -> k <> "=" <> v)         & key "process" . key "cwd" .~ toJSON (config & workingDirectory)+        & key "process" . key "user" . key "uid" .~ toJSON (virtualUID config)+        & key "process" . key "user" . key "gid" .~ toJSON (virtualGID config)+        & key "process" . key "user" . key "umask" .~ toJSON (0o077 :: Int)+        & key "process" . key "user" . key "additionalGids" . _Array .~ V.fromList []+        & key "linux" . _Object . at "uidMappings"+          ?~ toJSON+            [ object+                [ ("containerID", toJSON (virtualUID config)),+                  ("hostID", toJSON (hostUID config)),+                  ("size", toJSON (1 :: Int))+                ]+            ]+        & key "linux" . _Object . at "gidMappings"+          ?~ toJSON+            [ object+                [ ("containerID", toJSON (virtualGID config)),+                  ("hostID", toJSON (hostGID config)),+                  ("size", toJSON (1 :: Int))+                ]+            ]         & key "hostname" .~ toJSON (config & hostname)         & key "root" . key "readonly" .~ toJSON (config & rootReadOnly)+        -- TODO Use slirp? e.g. https://github.com/rootless-containers/slirp4netns or might kernel offer bridging (in the future?)+        & key "linux" . key "namespaces" . _Array %~ V.filter (\x -> x ^? key "type" . _String /= Just "network")  run :: FilePath -> Config -> IO ExitCode run dir config = do-  let runcExe = "runc"+  let containerRuntimeExe = "crun"       createConfigJsonSpec =-        (System.Process.proc runcExe ["spec", "--rootless"])+        (System.Process.proc containerRuntimeExe ["spec", "--rootless"])           { cwd = Just dir           }       configJsonPath = dir </> "config.json"-      runcRootPath = dir </> "runc-root"-      -- Although runc run --root says-      --    root directory for storage of container state (this should be located in tmpfs)-      -- this is not a requirement. See https://github.com/opencontainers/runc/issues/2054+      runtimeRootPath = dir </> "container-root"       rootfsPath = dir </> "rootfs"   (exit, _out, err) <- readCreateProcessWithExitCode createConfigJsonSpec ""   case exit of     ExitSuccess -> pass     ExitFailure e -> do       putErrText (decodeUtf8With lenientDecode err)-      panic $ "Could not create container configuration template. runc terminated with exit code " <> show e+      panic $ "Could not create container configuration template. " <> toS containerRuntimeExe <> " terminated with exit code " <> show e   templateBytes <- BS.readFile configJsonPath   template <- case eitherDecode (BL.fromStrict templateBytes) of     Right a -> pure a-    Left e -> throwIO (FatalError $ "decoding runc config.json template: " <> show e)-  let configJson = effectToRuncSpec config template+    Left e -> throwIO (FatalError $ "decoding container config.json template: " <> show e)+  let configJson = effectToOCIRuntimeSpec config template   BS.writeFile configJsonPath (BL.toStrict $ encode configJson)   createDirectory rootfsPath-  createDirectory runcRootPath+  createDirectory runtimeRootPath   name <- do     uuid <- UUID.nextRandom     pure $ "hercules-ci-" <> show uuid@@ -104,7 +128,7 @@       concurrently         ( do             let createProcSpec =-                  (System.Process.proc runcExe ["--root", runcRootPath, "run", name])+                  (System.Process.proc containerRuntimeExe ["--root", runtimeRootPath, "run", name])                     { std_in = UseHandle terminal, -- can't pass /dev/null :(                       std_out = UseHandle terminal,                       std_err = UseHandle terminal,@@ -114,10 +138,10 @@               waitForProcess processHandle                 `onException` ( do                                   putErrText "Terminating effect process..."-                                  _ <- System.Process.withCreateProcess (System.Process.proc runcExe ["kill", name]) \_ _ _ kh ->+                                  _ <- System.Process.withCreateProcess (System.Process.proc containerRuntimeExe ["kill", name]) \_ _ _ kh ->                                     waitForProcess kh                                   threadDelay 3_000_000-                                  _ <- System.Process.withCreateProcess (System.Process.proc runcExe ["kill", name, "KILL"]) \_ _ _ kh ->+                                  _ <- System.Process.withCreateProcess (System.Process.proc containerRuntimeExe ["kill", name, "KILL"]) \_ _ _ kh ->                                     waitForProcess kh                                   putErrText "Killed effect process."                               )

@@ -80,6 +80,9 @@       if actual == expect         then pure True         else False <$ tell ["isOwner: owner " <> show actual <> " is not the desired " <> show expect]+    eval (Hercules.Formats.Secret.Const b) = do+      tell ["const: " <> show b]+      pure b  -- This uses tagless final to derive both an efficient and a tracing function. 

@@ -0,0 +1,85 @@+{-# LANGUAGE BlockArguments #-}++module Hercules.Agent.ConduitSpec where++import Data.Conduit+import Data.Conduit.Combinators (sinkList)+import Data.Conduit.List (sourceList)+import Hercules.Agent.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]

@@ -0,0 +1,236 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# OPTIONS_GHC -Wno-orphans #-}++module Hercules.Agent.ConfigSpec (spec) where++import Data.Aeson qualified as A+import Data.Map qualified as M+import Data.Text qualified as T+import Hercules.Agent.Config (Config (..), Purpose (Input), combiCodec)+import Hercules.Agent.Config.Combined (forJson, forToml)+import Hercules.Agent.Config.Json qualified as Json+import Hercules.CNix.Verbosity (Verbosity (Vomit))+import Hercules.Formats.Mountable (Mountable (Mountable))+import Hercules.Formats.Mountable qualified+import Hercules.Formats.Secret qualified+import Katip (Severity (DebugS))+import Protolude+import Test.Hspec+import Toml.Codec.Code++deriving instance Eq (Config 'Input)++deriving instance Show (Config 'Input)++spec :: Spec+spec =+  describe "Config" $ do+    describe "toml codec" $ do+      -- note that this hooks into a more strict version of the parser+      it "parsers example 1 strictly" $ do+        let input =+              T.unlines+                [ "allowInsecureBuiltinFetchers = true",+                  "apiBaseUrl = \"http://api\"",+                  "baseDirectory = \"/var/lib/hercules-ci-agent\"",+                  "binaryCachesPath = \"/nix/store/9cbyq6f2ajxm64rvffhy8ndi73mbg5zd-binary-caches.json\"",+                  "clusterJoinTokenPath = \"/nix/store/y0200pr82sczzla5jd4pzkb8idnqhxkj-pretend-agent-token\"",+                  "concurrentTasks = 4",+                  "logLevel = \"DebugS\"",+                  "nixUserIsTrusted = true",+                  "secretsJsonPath = \"/var/lib/hercules-ci-agent/secrets/secrets.json\"",+                  "staticSecretsDirectory = \"/var/lib/hercules-ci-agent/secrets\"",+                  "workDirectory = \"/var/lib/hercules-ci-agent/work\"",+                  "remotePlatformsWithSameFeatures = [\"aarch64-darwin\"]",+                  "nixVerbosity = \"vomit\"",+                  -- This line should not be needed!+                  "[effectMountables]",+                  "[effectMountables.hosts]",+                  "  condition = true",+                  "  readOnly = true",+                  "  source = \"/etc/hosts\"",+                  "[labels]",+                  "module = \"nixos-profile\"",+                  "[labels.agent]",+                  "source = \"flake\"",+                  "[labels.lib]",+                  "version = \"24.05pre-git\""+                ]+        decode (forToml combiCodec) input+          `shouldBe` Right+            ( Config+                { herculesApiBaseURL = Just "http://api",+                  nixUserIsTrusted = Just True,+                  concurrentTasks = Just (Right 4),+                  baseDirectory = Just "/var/lib/hercules-ci-agent",+                  staticSecretsDirectory = Just "/var/lib/hercules-ci-agent/secrets",+                  workDirectory = Just "/var/lib/hercules-ci-agent/work",+                  clusterJoinTokenPath = Just "/nix/store/y0200pr82sczzla5jd4pzkb8idnqhxkj-pretend-agent-token",+                  binaryCachesPath = Just "/nix/store/9cbyq6f2ajxm64rvffhy8ndi73mbg5zd-binary-caches.json",+                  secretsJsonPath = Just "/var/lib/hercules-ci-agent/secrets/secrets.json",+                  logLevel = Just DebugS,+                  nixVerbosity = Nothing, -- FIXME: should be Just "vomit"+                  labels =+                    Just+                      ( M.fromList+                          [ ("agent", (A.object [("source", A.String "flake")])),+                            ("lib", A.object ([("version", A.String "24.05pre-git")])),+                            ("module", A.String "nixos-profile")+                          ]+                      ),+                  allowInsecureBuiltinFetchers = Just True,+                  remotePlatformsWithSameFeatures = Just ["aarch64-darwin"],+                  effectMountables =+                    M.fromList+                      [ ( "hosts",+                          Mountable+                            { source = "/etc/hosts",+                              readOnly = True,+                              condition = Hercules.Formats.Secret.Const True+                            }+                        )+                      ]+                }+            )+      it "parses empty config" $ do+        let input = T.unlines []+        decode (forToml combiCodec) input+          `shouldBe` Right+            emptyConfig+              { -- sure, why not+                labels = Just mempty+              }++    describe "json codec" $ do+      it "parses example 1" $ do+        let input =+              T.unlines+                [ "{",+                  "\"apiBaseUrl\": \"http://api\",",+                  "\"nixUserIsTrusted\": true,",+                  "\"concurrentTasks\": 4,",+                  "\"baseDirectory\": \"/var/lib/hercules-ci-agent\",",+                  "\"staticSecretsDirectory\": \"/var/lib/hercules-ci-agent/secrets\",",+                  "\"workDirectory\": \"/var/lib/hercules-ci-agent/work\",",+                  "\"clusterJoinTokenPath\": \"/nix/store/y0200pr82sczzla5jd4pzkb8idnqhxkj-pretend-agent-token\",",+                  "\"binaryCachesPath\": \"/nix/store/9cbyq6f2ajxm64rvffhy8ndi73mbg5zd-binary-caches.json\",",+                  "\"secretsJsonPath\": \"/var/lib/hercules-ci-agent/secrets/secrets.json\",",+                  "\"logLevel\": \"DebugS\",",+                  "\"nixVerbosity\": \"Vomit\",",+                  "\"remotePlatformsWithSameFeatures\": [\"aarch64-darwin\"],",+                  "\"labels\": {",+                  "  \"agent\": {\"source\": \"flake\"},",+                  "  \"lib\": {\"version\": \"24.05pre-git\"},",+                  "  \"module\": \"nixos-profile\"",+                  "},",+                  "\"effectMountables\": {",+                  "  \"hosts\": {",+                  "    \"source\": \"/etc/hosts\",",+                  "    \"readOnly\": true,",+                  "    \"condition\": true,",+                  "    \"and\":[]",+                  "  }",+                  "},",+                  "\"allowInsecureBuiltinFetchers\": true",+                  "}"+                ]+        Json.decode (forJson combiCodec) (toS $ encodeUtf8 input)+          `shouldBe` Right+            Config+              { herculesApiBaseURL = Just "http://api",+                nixUserIsTrusted = Just True,+                concurrentTasks = Just (Right 4),+                baseDirectory = Just "/var/lib/hercules-ci-agent",+                staticSecretsDirectory = Just "/var/lib/hercules-ci-agent/secrets",+                workDirectory = Just "/var/lib/hercules-ci-agent/work",+                clusterJoinTokenPath = Just "/nix/store/y0200pr82sczzla5jd4pzkb8idnqhxkj-pretend-agent-token",+                binaryCachesPath = Just "/nix/store/9cbyq6f2ajxm64rvffhy8ndi73mbg5zd-binary-caches.json",+                secretsJsonPath = Just "/var/lib/hercules-ci-agent/secrets/secrets.json",+                logLevel = Just DebugS,+                nixVerbosity = Just Vomit,+                labels =+                  Just+                    ( M.fromList+                        [ ("agent", (A.object [("source", A.String "flake")])),+                          ("lib", A.object ([("version", A.String "24.05pre-git")])),+                          ("module", A.String "nixos-profile")+                        ]+                    ),+                allowInsecureBuiltinFetchers = Just True,+                remotePlatformsWithSameFeatures = Just ["aarch64-darwin"],+                effectMountables =+                  M.fromList+                    [ ( "hosts",+                        Mountable+                          { source = "/etc/hosts",+                            readOnly = True,+                            condition = Hercules.Formats.Secret.Const True+                          }+                      )+                    ]+              }++      it "handles empty config" $ do+        let inputs =+              T.unlines+                [ "{",+                  "}"+                ]+        Json.decode (forJson combiCodec) (toS $ encodeUtf8 inputs) `shouldBe` Right emptyConfig++      it "handles nulls" $ do+        let inputs =+              T.unlines+                [ "{",+                  "\"apiBaseUrl\": null,",+                  "\"nixUserIsTrusted\": null,",+                  "\"concurrentTasks\": null,",+                  "\"baseDirectory\": null,",+                  "\"staticSecretsDirectory\": null,",+                  "\"workDirectory\": null,",+                  "\"clusterJoinTokenPath\": null,",+                  "\"binaryCachesPath\": null,",+                  "\"secretsJsonPath\": null,",+                  "\"logLevel\": null,",+                  "\"labels\": null,",+                  "\"allowInsecureBuiltinFetchers\": null",+                  "}"+                ]+        Json.decode (forJson combiCodec) (toS $ encodeUtf8 inputs) `shouldBe` Right emptyConfig++      it "allows use of null in labels" $ do+        let inputs =+              T.unlines+                [ "{",+                  "\"labels\": {",+                  "  \"version\": null",+                  "}",+                  "}"+                ]+        Json.decode (forJson combiCodec) (toS $ encodeUtf8 inputs)+          `shouldBe` Right+            emptyConfig+              { labels = Just (M.singleton "version" A.Null)+              }++emptyConfig :: Config 'Input+emptyConfig =+  Config+    { herculesApiBaseURL = Nothing,+      nixUserIsTrusted = Nothing,+      concurrentTasks = Nothing,+      baseDirectory = Nothing,+      staticSecretsDirectory = Nothing,+      workDirectory = Nothing,+      clusterJoinTokenPath = Nothing,+      binaryCachesPath = Nothing,+      secretsJsonPath = Nothing,+      logLevel = Nothing,+      nixVerbosity = Nothing,+      labels = Nothing,+      allowInsecureBuiltinFetchers = Nothing,+      remotePlatformsWithSameFeatures = Nothing,+      effectMountables = mempty+    }

@@ -1,85 +0,0 @@-{-# 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]

@@ -0,0 +1,28 @@+{-# LANGUAGE BlockArguments #-}++module Hercules.Agent.WorkerProtocol.WorkerConfigSpec where++import Data.Aeson qualified as A+import Hercules.Agent.WorkerProtocol.WorkerConfig+import Protolude+import Test.HUnit (assertFailure)+import Test.Hspec+import Test.QuickCheck++spec :: Spec+spec = do+  describe "WorkerConfig" do+    it "can be serialized and deserialized" do+      pending+  describe "ViaShowRead" do+    it "roundtrips aeson for Int" do+      property \i -> do+        let v = ViaShowRead (i :: Int)+        Just v === (A.decode . A.encode) v+    it "fails for invalid input" do+      e <- shouldBeLeft $ A.eitherDecode @(ViaShowRead Int) "\"invalid\""+      e `shouldContain` "Could not parse \"invalid\""++shouldBeLeft :: forall b a. Either a b -> IO a+shouldBeLeft (Left a) = pure a+shouldBeLeft (Right _) = assertFailure "Expected Left, got Right"