libmain: add progress bar with multiple status lines

Add the log-formats `multiline` and `multiline-with-logs` which offer
multiple current active building status lines.

Change-Id: Idd8afe62f8591b5d8b70e258c5cefa09be4cab03

doc/manual/src/command-ref/opt-common.md

@@ -87,7 +87,6 @@ Most commands in Lix accept the following command-line options:
 
     Displayes the raw logs, with a progress bar and activities each in a new line at the bottom.
 
-
 - <span id="opt-no-build-output">[`--no-build-output`](#opt-no-build-output)</span> / `-Q`
 
   By default, output written by builders to standard output and standard error is echoed to the Lix command's standard error.

src/libcmd/command-installable-value.cc

@@ -0,0 +1,11 @@
+#include "command-installable-value.hh"
+
+namespace nix {
+
+void InstallableValueCommand::run(ref<Store> store, ref<Installable> installable)
+{
+    auto installableValue = InstallableValue::require(installable);
+    run(store, installableValue);
+}
+
+}

src/libcmd/command-installable-value.hh

@@ -0,0 +1,23 @@
+#pragma once
+///@file
+
+#include "installable-value.hh"
+#include "command.hh"
+
+namespace nix {
+
+/**
+ * An InstallableCommand where the single positional argument must be an
+ * InstallableValue in particular.
+ */
+struct InstallableValueCommand : InstallableCommand
+{
+    /**
+     * Entry point to this command
+     */
+    virtual void run(ref<Store> store, ref<InstallableValue> installable) = 0;
+
+    void run(ref<Store> store, ref<Installable> installable) override;
+};
+
+}

src/libcmd/installables.cc

@@ -393,10 +393,13 @@ ref<eval_cache::EvalCache> openEvalCache(
     EvalState & state,
     std::shared_ptr<flake::LockedFlake> lockedFlake)
 {
-    auto fingerprint = evalSettings.useEvalCache && evalSettings.pureEval
-        ? std::make_optional(lockedFlake->getFingerprint())
-        : std::nullopt;
-    auto rootLoader = [&state, lockedFlake]()
+    auto fingerprint = lockedFlake->getFingerprint();
+    return make_ref<nix::eval_cache::EvalCache>(
+        evalSettings.useEvalCache && evalSettings.pureEval
+            ? std::optional { std::cref(fingerprint) }
+            : std::nullopt,
+        state,
+        [&state, lockedFlake]()
         {
             /* For testing whether the evaluation cache is
                complete. */
@@ -412,17 +415,7 @@ ref<eval_cache::EvalCache> openEvalCache(
             assert(aOutputs);
 
             return aOutputs->value;
-        };
-
-    if (fingerprint) {
-        auto search = state.evalCaches.find(fingerprint.value());
-        if (search == state.evalCaches.end()) {
-            search = state.evalCaches.emplace(fingerprint.value(), make_ref<nix::eval_cache::EvalCache>(fingerprint, state, rootLoader)).first;
-        }
-        return search->second;
-    } else {
-        return make_ref<nix::eval_cache::EvalCache>(std::nullopt, state, rootLoader);
-    }
+        });
 }
 
 Installables SourceExprCommand::parseInstallables(

src/libcmd/meson.build

@@ -1,5 +1,6 @@
 libcmd_sources = files(
   'built-path.cc',
+  'command-installable-value.cc',
   'cmd-profiles.cc',
   'command.cc',
   'common-eval-args.cc',
@@ -17,6 +18,7 @@ libcmd_sources = files(
 
 libcmd_headers = files(
   'built-path.hh',
+  'command-installable-value.hh',
   'cmd-profiles.hh',
   'command.hh',
   'common-eval-args.hh',

src/libexpr/eval.hh

@@ -33,10 +33,6 @@ class EvalState;
 class StorePath;
 struct SingleDerivedPath;
 enum RepairFlag : bool;
-struct MemoryInputAccessor;
-namespace eval_cache {
-    class EvalCache;
-}
 
 
 /**
@@ -238,11 +234,6 @@ public:
         return *new EvalErrorBuilder<T>(*this, args...);
     }
 
-    /**
-     * A cache for evaluation caches, so as to reuse the same root value if possible
-     */
-    std::map<const Hash, ref<eval_cache::EvalCache>> evalCaches;
-
 private:
 
     /* Cache for calls to addToStore(); maps source paths to the store

src/libmain/progress-bar.cc

@@ -36,475 +36,547 @@ static std::string_view storePathToName(std::string_view path)
     return i == std::string::npos ? base.substr(0, 0) : base.substr(i + 1);
 }
 
+// 100 years ought to be enough for anyone (yet sufficiently smaller than max() to not cause signed integer overflow).
+constexpr const auto A_LONG_TIME = std::chrono::duration_cast<std::chrono::milliseconds>(100 * 365 * 86400s);
 
-ProgressBar::~ProgressBar()
+class ProgressBar : public Logger
 {
-    stop();
-}
+private:
 
-/* Called by destructor, can't be overridden */
-void ProgressBar::stop()
-{
+    struct ActInfo
+    {
+        std::string s, lastLine, phase;
+        ActivityType type = actUnknown;
+        uint64_t done = 0;
+        uint64_t expected = 0;
+        uint64_t running = 0;
+        uint64_t failed = 0;
+        std::map<ActivityType, uint64_t> expectedByType;
+        bool visible = true;
+        ActivityId parent;
+        std::optional<std::string> name;
+        std::chrono::time_point<std::chrono::steady_clock> startTime;
+    };
+
+    struct ActivitiesByType
+    {
+        std::map<ActivityId, std::list<ActInfo>::iterator> its;
+        uint64_t done = 0;
+        uint64_t expected = 0;
+        uint64_t failed = 0;
+    };
+
+    struct State
+    {
+        std::list<ActInfo> activities;
+        std::map<ActivityId, std::list<ActInfo>::iterator> its;
+
+        std::map<ActivityType, ActivitiesByType> activitiesByType;
+
+        int lastLines = 0;
+
+        uint64_t filesLinked = 0, bytesLinked = 0;
+
+        uint64_t corruptedPaths = 0, untrustedPaths = 0;
+
+        bool active = true;
+        bool paused = false;
+        bool haveUpdate = true;
+    };
+
+    Sync<State> state_;
+
+    std::thread updateThread;
+
+    std::condition_variable quitCV, updateCV;
+
+    bool printBuildLogs = false;
+    bool printMultiline = false;
+    bool isTTY;
+
+public:
+
+    ProgressBar(bool isTTY)
+        : isTTY(isTTY)
+    {
+        state_.lock()->active = isTTY;
+        updateThread = std::thread([&]() {
+            auto state(state_.lock());
+            auto nextWakeup = A_LONG_TIME;
+            while (state->active) {
+                if (!state->haveUpdate)
+                    state.wait_for(updateCV, nextWakeup);
+                nextWakeup = draw(*state, {});
+                state.wait_for(quitCV, std::chrono::milliseconds(50));
+            }
+        });
+    }
+
+    ~ProgressBar()
+    {
+        stop();
+    }
+
+    /* Called by destructor, can't be overridden */
+    void stop() override final
+    {
+        {
+            auto state(state_.lock());
+            if (!state->active) return;
+            state->active = false;
+            writeToStderr("\r\e[K");
+            updateCV.notify_one();
+            quitCV.notify_one();
+        }
+        updateThread.join();
+    }
+
+    void pause() override {
+        state_.lock()->paused = true;
+        writeToStderr("\r\e[K");
+    }
+
+    void resume() override {
+        state_.lock()->paused = false;
+        writeToStderr("\r\e[K");
+        state_.lock()->haveUpdate = true;
+        updateCV.notify_one();
+    }
+
+    bool isVerbose() override
+    {
+        return printBuildLogs;
+    }
+
+    void log(Verbosity lvl, std::string_view s) override
+    {
+        if (lvl > verbosity) return;
+        auto state(state_.lock());
+        log(*state, lvl, s);
+    }
+
+    void logEI(const ErrorInfo & ei) override
     {
         auto state(state_.lock());
-        if (!state->active) return;
-        state->active = false;
-        writeToStderr("\r\e[K");
-        updateCV.notify_one();
-        quitCV.notify_one();
+
+        std::stringstream oss;
+        showErrorInfo(oss, ei, loggerSettings.showTrace.get());
+
+        log(*state, ei.level, oss.str());
     }
-    updateThread.join();
-}
 
-void ProgressBar::pause()
-{
-    state_.lock()->paused = true;
-    writeToStderr("\r\e[K");
-}
-
-void ProgressBar::resume()
-{
-    state_.lock()->paused = false;
-    writeToStderr("\r\e[K");
-    state_.lock()->haveUpdate = true;
-    updateCV.notify_one();
-}
-
-bool ProgressBar::isVerbose()
-{
-    return printBuildLogs;
-}
-
-void ProgressBar::log(Verbosity lvl, std::string_view s)
-{
-    if (lvl > verbosity) return;
-    auto state(state_.lock());
-    log(*state, lvl, s);
-}
-
-void ProgressBar::logEI(const ErrorInfo & ei)
-{
-    auto state(state_.lock());
-
-    std::stringstream oss;
-    showErrorInfo(oss, ei, loggerSettings.showTrace.get());
-
-    log(*state, ei.level, oss.str());
-}
-
-void ProgressBar::log(State & state, Verbosity lvl, std::string_view s)
-{
-    if (state.active) {
-        draw(state, s);
-    } else {
-        auto s2 = s + ANSI_NORMAL "\n";
-        if (!isTTY) s2 = filterANSIEscapes(s2, true);
-        writeToStderr(s2);
-    }
-}
-
-void ProgressBar::startActivity(
-    ActivityId act,
-    Verbosity lvl,
-    ActivityType type,
-    const std::string & s,
-    const Fields & fields,
-    ActivityId parent
-)
-{
-    auto state(state_.lock());
-
-    if (lvl <= verbosity && !s.empty() && type != actBuildWaiting)
-        log(*state, lvl, s + "...");
-
-    state->activities.emplace_back(ActInfo {
-        .s = s,
-        .type = type,
-        .parent = parent,
-        .startTime = std::chrono::steady_clock::now()
-    });
-    auto i = std::prev(state->activities.end());
-    state->its.emplace(act, i);
-    state->activitiesByType[type].its.emplace(act, i);
-
-    if (type == actBuild) {
-        std::string name(storePathToName(getS(fields, 0)));
-        if (name.ends_with(".drv"))
-            name = name.substr(0, name.size() - 4);
-        i->s = fmt("building " ANSI_BOLD "%s" ANSI_NORMAL, name);
-        auto machineName = getS(fields, 1);
-        if (machineName != "")
-            i->s += fmt(" on " ANSI_BOLD "%s" ANSI_NORMAL, machineName);
-
-        // Used to be curRound and nrRounds, but the
-        // implementation was broken for a long time.
-        if (getI(fields, 2) != 1 || getI(fields, 3) != 1) {
-            throw Error("log message indicated repeating builds, but this is not currently implemented");
+    void log(State & state, Verbosity lvl, std::string_view s)
+    {
+        if (state.active) {
+            draw(state, s);
+        } else {
+            auto s2 = s + ANSI_NORMAL "\n";
+            if (!isTTY) s2 = filterANSIEscapes(s2, true);
+            writeToStderr(s2);
         }
-        i->name = DrvName(name).name;
     }
 
-    if (type == actSubstitute) {
-        auto name = storePathToName(getS(fields, 0));
-        auto sub = getS(fields, 1);
-        i->s = fmt(
-            sub.starts_with("local")
-            ? "copying " ANSI_BOLD "%s" ANSI_NORMAL " from %s"
-            : "fetching " ANSI_BOLD "%s" ANSI_NORMAL " from %s",
-            name, sub);
-    }
+    void startActivity(ActivityId act, Verbosity lvl, ActivityType type,
+        const std::string & s, const Fields & fields, ActivityId parent) override
+    {
+        auto state(state_.lock());
 
-    if (type == actPostBuildHook) {
-        auto name = storePathToName(getS(fields, 0));
-        if (name.ends_with(".drv"))
-            name = name.substr(0, name.size() - 4);
-        i->s = fmt("post-build " ANSI_BOLD "%s" ANSI_NORMAL, name);
-        i->name = DrvName(name).name;
-    }
+        if (lvl <= verbosity && !s.empty() && type != actBuildWaiting)
+            log(*state, lvl, s + "...");
 
-    if (type == actQueryPathInfo) {
-        auto name = storePathToName(getS(fields, 0));
-        i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1));
-    }
+        state->activities.emplace_back(ActInfo {
+            .s = s,
+            .type = type,
+            .parent = parent,
+            .startTime = std::chrono::steady_clock::now()
+        });
+        auto i = std::prev(state->activities.end());
+        state->its.emplace(act, i);
+        state->activitiesByType[type].its.emplace(act, i);
 
-    if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent))
-        || (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent))
-        || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)))
-        i->visible = false;
+        if (type == actBuild) {
+            std::string name(storePathToName(getS(fields, 0)));
+            if (name.ends_with(".drv"))
+                name = name.substr(0, name.size() - 4);
+            i->s = fmt("building " ANSI_BOLD "%s" ANSI_NORMAL, name);
+            auto machineName = getS(fields, 1);
+            if (machineName != "")
+                i->s += fmt(" on " ANSI_BOLD "%s" ANSI_NORMAL, machineName);
 
-    update(*state);
-}
+            // Used to be curRound and nrRounds, but the
+            // implementation was broken for a long time.
+            if (getI(fields, 2) != 1 || getI(fields, 3) != 1) {
+                throw Error("log message indicated repeating builds, but this is not currently implemented");
+            }
+            i->name = DrvName(name).name;
+        }
 
-/* Check whether an activity has an ancestore with the specified
-   type. */
-bool ProgressBar::hasAncestor(State & state, ActivityType type, ActivityId act)
-{
-    while (act != 0) {
-        auto i = state.its.find(act);
-        if (i == state.its.end()) break;
-        if (i->second->type == type) return true;
-        act = i->second->parent;
-    }
-    return false;
-}
+        if (type == actSubstitute) {
+            auto name = storePathToName(getS(fields, 0));
+            auto sub = getS(fields, 1);
+            i->s = fmt(
+                sub.starts_with("local")
+                ? "copying " ANSI_BOLD "%s" ANSI_NORMAL " from %s"
+                : "fetching " ANSI_BOLD "%s" ANSI_NORMAL " from %s",
+                name, sub);
+        }
 
-void ProgressBar::stopActivity(ActivityId act)
-{
-    auto state(state_.lock());
+        if (type == actPostBuildHook) {
+            auto name = storePathToName(getS(fields, 0));
+            if (name.ends_with(".drv"))
+                name = name.substr(0, name.size() - 4);
+            i->s = fmt("post-build " ANSI_BOLD "%s" ANSI_NORMAL, name);
+            i->name = DrvName(name).name;
+        }
 
-    auto i = state->its.find(act);
-    if (i != state->its.end()) {
+        if (type == actQueryPathInfo) {
+            auto name = storePathToName(getS(fields, 0));
+            i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1));
+        }
 
-        auto & actByType = state->activitiesByType[i->second->type];
-        actByType.done += i->second->done;
-        actByType.failed += i->second->failed;
+        if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent))
+            || (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent))
+            || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)))
+            i->visible = false;
 
-        for (auto & j : i->second->expectedByType)
-            state->activitiesByType[j.first].expected -= j.second;
-
-        actByType.its.erase(act);
-        state->activities.erase(i->second);
-        state->its.erase(i);
-    }
-
-    update(*state);
-}
-
-void ProgressBar::result(ActivityId act, ResultType type, const std::vector<Field> & fields)
-{
-    auto state(state_.lock());
-
-    if (type == resFileLinked) {
-        state->filesLinked++;
-        state->bytesLinked += getI(fields, 0);
         update(*state);
     }
 
-    else if (type == resBuildLogLine || type == resPostBuildLogLine) {
-        auto lastLine = chomp(getS(fields, 0));
-        if (!lastLine.empty()) {
+    /* Check whether an activity has an ancestore with the specified
+       type. */
+    bool hasAncestor(State & state, ActivityType type, ActivityId act)
+    {
+        while (act != 0) {
+            auto i = state.its.find(act);
+            if (i == state.its.end()) break;
+            if (i->second->type == type) return true;
+            act = i->second->parent;
+        }
+        return false;
+    }
+
+    void stopActivity(ActivityId act) override
+    {
+        auto state(state_.lock());
+
+        auto i = state->its.find(act);
+        if (i != state->its.end()) {
+
+            auto & actByType = state->activitiesByType[i->second->type];
+            actByType.done += i->second->done;
+            actByType.failed += i->second->failed;
+
+            for (auto & j : i->second->expectedByType)
+                state->activitiesByType[j.first].expected -= j.second;
+
+            actByType.its.erase(act);
+            state->activities.erase(i->second);
+            state->its.erase(i);
+        }
+
+        update(*state);
+    }
+
+    void result(ActivityId act, ResultType type, const std::vector<Field> & fields) override
+    {
+        auto state(state_.lock());
+
+        if (type == resFileLinked) {
+            state->filesLinked++;
+            state->bytesLinked += getI(fields, 0);
+            update(*state);
+        }
+
+        else if (type == resBuildLogLine || type == resPostBuildLogLine) {
+            auto lastLine = chomp(getS(fields, 0));
+            if (!lastLine.empty()) {
+                auto i = state->its.find(act);
+                assert(i != state->its.end());
+                ActInfo info = *i->second;
+                if (printBuildLogs) {
+                    auto suffix = "> ";
+                    if (type == resPostBuildLogLine) {
+                        suffix = " (post)> ";
+                    }
+                    log(*state, lvlInfo, ANSI_FAINT + info.name.value_or("unnamed") + suffix + ANSI_NORMAL + lastLine);
+                } else {
+                    if (!printMultiline) {
+                        state->activities.erase(i->second);
+                        info.lastLine = lastLine;
+                        state->activities.emplace_back(info);
+                        i->second = std::prev(state->activities.end());
+                    } else {
+                        i->second->lastLine = lastLine;
+                    }
+                    update(*state);
+                }
+            }
+        }
+
+        else if (type == resUntrustedPath) {
+            state->untrustedPaths++;
+            update(*state);
+        }
+
+        else if (type == resCorruptedPath) {
+            state->corruptedPaths++;
+            update(*state);
+        }
+
+        else if (type == resSetPhase) {
             auto i = state->its.find(act);
             assert(i != state->its.end());
-            ActInfo info = *i->second;
-            if (printBuildLogs) {
-                auto suffix = "> ";
-                if (type == resPostBuildLogLine) {
-                    suffix = " (post)> ";
-                }
-                log(*state, lvlInfo, ANSI_FAINT + info.name.value_or("unnamed") + suffix + ANSI_NORMAL + lastLine);
-            } else {
-                if (!printMultiline) {
-                    state->activities.erase(i->second);
-                    info.lastLine = lastLine;
-                    state->activities.emplace_back(info);
-                    i->second = std::prev(state->activities.end());
-                } else {
-                    i->second->lastLine = lastLine;
-                }
-                update(*state);
-            }
+            i->second->phase = getS(fields, 0);
+            update(*state);
+        }
+
+        else if (type == resProgress) {
+            auto i = state->its.find(act);
+            assert(i != state->its.end());
+            ActInfo & actInfo = *i->second;
+            actInfo.done = getI(fields, 0);
+            actInfo.expected = getI(fields, 1);
+            actInfo.running = getI(fields, 2);
+            actInfo.failed = getI(fields, 3);
+            update(*state);
+        }
+
+        else if (type == resSetExpected) {
+            auto i = state->its.find(act);
+            assert(i != state->its.end());
+            ActInfo & actInfo = *i->second;
+            auto type = (ActivityType) getI(fields, 0);
+            auto & j = actInfo.expectedByType[type];
+            state->activitiesByType[type].expected -= j;
+            j = getI(fields, 1);
+            state->activitiesByType[type].expected += j;
+            update(*state);
         }
     }
 
-    else if (type == resUntrustedPath) {
-        state->untrustedPaths++;
-        update(*state);
+    void update(State & state)
+    {
+        state.haveUpdate = true;
+        updateCV.notify_one();
     }
 
-    else if (type == resCorruptedPath) {
-        state->corruptedPaths++;
-        update(*state);
-    }
+    std::chrono::milliseconds draw(State & state, const std::optional<std::string_view> & s)
+    {
+        auto nextWakeup = A_LONG_TIME;
 
-    else if (type == resSetPhase) {
-        auto i = state->its.find(act);
-        assert(i != state->its.end());
-        i->second->phase = getS(fields, 0);
-        update(*state);
-    }
+        state.haveUpdate = false;
+        if (state.paused || !state.active) return nextWakeup;
 
-    else if (type == resProgress) {
-        auto i = state->its.find(act);
-        assert(i != state->its.end());
-        ActInfo & actInfo = *i->second;
-        actInfo.done = getI(fields, 0);
-        actInfo.expected = getI(fields, 1);
-        actInfo.running = getI(fields, 2);
-        actInfo.failed = getI(fields, 3);
-        update(*state);
-    }
+        auto windowSize = getWindowSize();
+        auto width = windowSize.second;
+        if (width <= 0) {
+            width = std::numeric_limits<decltype(width)>::max();
+        }
 
-    else if (type == resSetExpected) {
-        auto i = state->its.find(act);
-        assert(i != state->its.end());
-        ActInfo & actInfo = *i->second;
-        auto type = (ActivityType) getI(fields, 0);
-        auto & j = actInfo.expectedByType[type];
-        state->activitiesByType[type].expected -= j;
-        j = getI(fields, 1);
-        state->activitiesByType[type].expected += j;
-        update(*state);
-    }
-}
+        if (printMultiline && (state.lastLines >= 1)) {
+            // FIXME: make sure this works on windows
+            writeToStderr(fmt("\e[G\e[%dF\e[J", state.lastLines));
+        }
 
-void ProgressBar::update(State & state)
-{
-    state.haveUpdate = true;
-    updateCV.notify_one();
-}
+        state.lastLines = 0;
 
-std::chrono::milliseconds ProgressBar::draw(State & state, const std::optional<std::string_view> & s)
-{
-    auto nextWakeup = A_LONG_TIME;
+        if (s != std::nullopt)
+            writeToStderr("\r\e[K" + filterANSIEscapes(s.value(), !isTTY) + ANSI_NORMAL "\n");
 
-    state.haveUpdate = false;
-    if (state.paused || !state.active) return nextWakeup;
+        std::string line;
+        std::string status = getStatus(state);
+        if (!status.empty()) {
+            line += '[';
+            line += status;
+            line += "]";
+        }
+        if (printMultiline && !line.empty()) {
+            writeToStderr(filterANSIEscapes(line, false, width) + "\n");
+            state.lastLines++;
+        }
 
-    auto windowSize = getWindowSize();
-    auto width = windowSize.second;
-    if (width <= 0) {
-        width = std::numeric_limits<decltype(width)>::max();
-    }
+        auto height = windowSize.first > 0 ? windowSize.first : 25;
+        auto moreBuilds = 0;
+        auto now = std::chrono::steady_clock::now();
 
-    if (printMultiline && (state.lastLines >= 1)) {
-        // FIXME: make sure this works on windows
-        writeToStderr(fmt("\e[G\e[%dF\e[J", state.lastLines));
-    }
+        std::string activity_line;
+        if (!state.activities.empty()) {
+            for (auto i = state.activities.begin(); i != state.activities.end(); ++i) {
+                if (!(i->visible && (!i->s.empty() || !i->lastLine.empty()))) {
+                    continue;
+                }
 
-    state.lastLines = 0;
-
-    if (s != std::nullopt)
-        writeToStderr("\r\e[K" + filterANSIEscapes(s.value(), !isTTY) + ANSI_NORMAL "\n");
-
-    std::string line;
-    std::string status = getStatus(state);
-    if (!status.empty()) {
-        line += '[';
-        line += status;
-        line += "]";
-    }
-    if (printMultiline && !line.empty()) {
-        writeToStderr(filterANSIEscapes(line, false, width) + "\n");
-        state.lastLines++;
-    }
-
-     auto height = windowSize.first > 0 ? windowSize.first : 25;
-     auto moreActivities = 0;
-    auto now = std::chrono::steady_clock::now();
-
-    std::string activity_line;
-    if (!state.activities.empty()) {
-        for (auto i = state.activities.begin(); i != state.activities.end(); ++i) {
-            if (!(i->visible && (!i->s.empty() || !i->lastLine.empty()))) {
-                continue;
-            }
-            /* Don't show activities until some time has
-               passed, to avoid displaying very short
-               activities. */
-            auto delay = std::chrono::milliseconds(10);
-            if (i->startTime + delay >= now) {
-                nextWakeup = std::min(
-                    nextWakeup,
-                    std::chrono::duration_cast<std::chrono::milliseconds>(
-                        delay - (now - i->startTime)
+                /* Don't show activities until some time has
+                   passed, to avoid displaying very short
+                   activities. */
+                auto delay = std::chrono::milliseconds(10);
+                if (i->startTime + delay >= now) {
+                    nextWakeup = std::min(
+                        nextWakeup,
+                        std::chrono::duration_cast<std::chrono::milliseconds>(
+                            delay - (now - i->startTime)
                         )
                     );
-            }
+                }
 
-            activity_line = i->s;
+                activity_line = i->s;
 
-            if (!i->phase.empty()) {
-                activity_line += " (";
-                activity_line += i->phase;
-                activity_line += ")";
-            }
-            if (!i->lastLine.empty()) {
-                if (!i->s.empty())
-                    activity_line += ": ";
-                activity_line += i->lastLine;
-            }
-
-            if (printMultiline) {
-                if (state.lastLines < (height -1)) {
-                    writeToStderr(filterANSIEscapes(activity_line, false, width) + "\n");
-                    state.lastLines++;
-                } else moreActivities++;
+                if (!i->phase.empty()) {
+                    activity_line += " (";
+                    activity_line += i->phase;
+                    activity_line += ")";
+                }
+                if (!i->lastLine.empty()) {
+                    if (!i->s.empty()) {
+                        activity_line += ": ";
+                    }
+                    activity_line += i->lastLine;
+                }
+                if (printMultiline) {
+                    if (state.lastLines < (height - 1)) {
+                        writeToStderr(filterANSIEscapes(activity_line, false, width) + "\n");
+                        state.lastLines++;
+                    } else {
+                        moreBuilds++;
+                    }
+                }
             }
         }
-    }
 
-    if (printMultiline && moreActivities)
-        writeToStderr(fmt("And %d more...", moreActivities));
-
-    if (!printMultiline) {
-        line += " " + activity_line;
-         writeToStderr("\r" + filterANSIEscapes(line, false, width) + ANSI_NORMAL + "\e[K");
-    }
-
-    return nextWakeup;
-}
-
-std::string ProgressBar::getStatus(State & state)
-{
-    constexpr auto MiB = 1024.0 * 1024.0;
-
-    std::string res;
-
-    auto renderActivity = [&](ActivityType type, const std::string & itemFmt, const std::string & numberFmt = "%d", double unit = 1) {
-        auto & act = state.activitiesByType[type];
-        uint64_t done = act.done, expected = act.done, running = 0, failed = act.failed;
-        for (auto & [actId, infoIt] : act.its) {
-            done += infoIt->done;
-            expected += infoIt->expected;
-            running += infoIt->running;
-            failed += infoIt->failed;
+        if (printMultiline && moreBuilds) {
+            writeToStderr(fmt("And %d more...", moreBuilds));
         }
 
-        expected = std::max(expected, act.expected);
-
-        std::string s;
-
-        if (running || done || expected || failed) {
-            if (running)
-                if (expected != 0)
-                    s = fmt(ANSI_BLUE + numberFmt + ANSI_NORMAL "/" ANSI_GREEN + numberFmt + ANSI_NORMAL "/" + numberFmt,
-                        running / unit, done / unit, expected / unit);
-                else
-                    s = fmt(ANSI_BLUE + numberFmt + ANSI_NORMAL "/" ANSI_GREEN + numberFmt + ANSI_NORMAL,
-                        running / unit, done / unit);
-            else if (expected != done)
-                if (expected != 0)
-                    s = fmt(ANSI_GREEN + numberFmt + ANSI_NORMAL "/" + numberFmt,
-                        done / unit, expected / unit);
-                else
-                    s = fmt(ANSI_GREEN + numberFmt + ANSI_NORMAL, done / unit);
-            else
-                s = fmt(done ? ANSI_GREEN + numberFmt + ANSI_NORMAL : numberFmt, done / unit);
-            s = fmt(itemFmt, s);
-
-            if (failed)
-                s += fmt(" (" ANSI_RED "%d failed" ANSI_NORMAL ")", failed / unit);
+        if (!printMultiline) {
+            line += " " + activity_line;
+            writeToStderr("\r" + filterANSIEscapes(line, false, width) + ANSI_NORMAL + "\e[K");
         }
 
-        return s;
-    };
-
-    auto showActivity = [&](ActivityType type, const std::string & itemFmt, const std::string & numberFmt = "%d", double unit = 1) {
-        auto s = renderActivity(type, itemFmt, numberFmt, unit);
-        if (s.empty()) return;
-        if (!res.empty()) res += ", ";
-        res += s;
-    };
-
-    showActivity(actBuilds, "%s built");
-
-    auto s1 = renderActivity(actCopyPaths, "%s copied");
-    auto s2 = renderActivity(actCopyPath, "%s MiB", "%.1f", MiB);
-
-    if (!s1.empty() || !s2.empty()) {
-        if (!res.empty()) res += ", ";
-        if (s1.empty()) res += "0 copied"; else res += s1;
-        if (!s2.empty()) { res += " ("; res += s2; res += ')'; }
+        return nextWakeup;
     }
 
-    showActivity(actFileTransfer, "%s MiB DL", "%.1f", MiB);
-
+    std::string getStatus(State & state)
     {
-        auto s = renderActivity(actOptimiseStore, "%s paths optimised");
-        if (s != "") {
-            s += fmt(", %.1f MiB / %d inodes freed", state.bytesLinked / MiB, state.filesLinked);
+        auto MiB = 1024.0 * 1024.0;
+
+        std::string res;
+
+        auto renderActivity = [&](ActivityType type, const std::string & itemFmt, const std::string & numberFmt = "%d", double unit = 1) {
+            auto & act = state.activitiesByType[type];
+            uint64_t done = act.done, expected = act.done, running = 0, failed = act.failed;
+            for (auto & j : act.its) {
+                done += j.second->done;
+                expected += j.second->expected;
+                running += j.second->running;
+                failed += j.second->failed;
+            }
+
+            expected = std::max(expected, act.expected);
+
+            std::string s;
+
+            if (running || done || expected || failed) {
+                if (running)
+                    if (expected != 0)
+                        s = fmt(ANSI_BLUE + numberFmt + ANSI_NORMAL "/" ANSI_GREEN + numberFmt + ANSI_NORMAL "/" + numberFmt,
+                            running / unit, done / unit, expected / unit);
+                    else
+                        s = fmt(ANSI_BLUE + numberFmt + ANSI_NORMAL "/" ANSI_GREEN + numberFmt + ANSI_NORMAL,
+                            running / unit, done / unit);
+                else if (expected != done)
+                    if (expected != 0)
+                        s = fmt(ANSI_GREEN + numberFmt + ANSI_NORMAL "/" + numberFmt,
+                            done / unit, expected / unit);
+                    else
+                        s = fmt(ANSI_GREEN + numberFmt + ANSI_NORMAL, done / unit);
+                else
+                    s = fmt(done ? ANSI_GREEN + numberFmt + ANSI_NORMAL : numberFmt, done / unit);
+                s = fmt(itemFmt, s);
+
+                if (failed)
+                    s += fmt(" (" ANSI_RED "%d failed" ANSI_NORMAL ")", failed / unit);
+            }
+
+            return s;
+        };
+
+        auto showActivity = [&](ActivityType type, const std::string & itemFmt, const std::string & numberFmt = "%d", double unit = 1) {
+            auto s = renderActivity(type, itemFmt, numberFmt, unit);
+            if (s.empty()) return;
             if (!res.empty()) res += ", ";
             res += s;
+        };
+
+        showActivity(actBuilds, "%s built");
+
+        auto s1 = renderActivity(actCopyPaths, "%s copied");
+        auto s2 = renderActivity(actCopyPath, "%s MiB", "%.1f", MiB);
+
+        if (!s1.empty() || !s2.empty()) {
+            if (!res.empty()) res += ", ";
+            if (s1.empty()) res += "0 copied"; else res += s1;
+            if (!s2.empty()) { res += " ("; res += s2; res += ')'; }
+        }
+
+        showActivity(actFileTransfer, "%s MiB DL", "%.1f", MiB);
+
+        {
+            auto s = renderActivity(actOptimiseStore, "%s paths optimised");
+            if (s != "") {
+                s += fmt(", %.1f MiB / %d inodes freed", state.bytesLinked / MiB, state.filesLinked);
+                if (!res.empty()) res += ", ";
+                res += s;
+            }
+        }
+
+        // FIXME: don't show "done" paths in green.
+        showActivity(actVerifyPaths, "%s paths verified");
+
+        if (state.corruptedPaths) {
+            if (!res.empty()) res += ", ";
+            res += fmt(ANSI_RED "%d corrupted" ANSI_NORMAL, state.corruptedPaths);
+        }
+
+        if (state.untrustedPaths) {
+            if (!res.empty()) res += ", ";
+            res += fmt(ANSI_RED "%d untrusted" ANSI_NORMAL, state.untrustedPaths);
+        }
+
+        return res;
+    }
+
+    void writeToStdout(std::string_view s) override
+    {
+        auto state(state_.lock());
+        if (state->active) {
+            Logger::writeToStdout(s);
+            draw(*state, {});
+        } else {
+            Logger::writeToStdout(s);
         }
     }
 
-    // FIXME: don't show "done" paths in green.
-    showActivity(actVerifyPaths, "%s paths verified");
-
-    if (state.corruptedPaths) {
-        if (!res.empty()) res += ", ";
-        res += fmt(ANSI_RED "%d corrupted" ANSI_NORMAL, state.corruptedPaths);
-    }
-
-    if (state.untrustedPaths) {
-        if (!res.empty()) res += ", ";
-        res += fmt(ANSI_RED "%d untrusted" ANSI_NORMAL, state.untrustedPaths);
-    }
-
-    return res;
-}
-
-void ProgressBar::writeToStdout(std::string_view s)
-{
-    auto state(state_.lock());
-    if (state->active) {
-        Logger::writeToStdout(s);
+    std::optional<char> ask(std::string_view msg) override
+    {
+        auto state(state_.lock());
+        if (!state->active || !isatty(STDIN_FILENO)) return {};
+        std::cerr << fmt("\r\e[K%s ", msg);
+        auto s = trim(readLine(STDIN_FILENO));
+        if (s.size() != 1) return {};
         draw(*state, {});
-    } else {
-        Logger::writeToStdout(s);
+        return s[0];
     }
-}
 
-std::optional<char> ProgressBar::ask(std::string_view msg)
-{
-    auto state(state_.lock());
-    if (!state->active || !isatty(STDIN_FILENO)) return {};
-    std::cerr << fmt("\r\e[K%s ", msg);
-    auto s = trim(readLine(STDIN_FILENO));
-    if (s.size() != 1) return {};
-    draw(*state, {});
-    return s[0];
-}
+    void setPrintBuildLogs(bool printBuildLogs) override
+    {
+        this->printBuildLogs = printBuildLogs;
+    }
 
-void ProgressBar::setPrintBuildLogs(bool printBuildLogs)
-{
-    this->printBuildLogs = printBuildLogs;
-}
-
-void ProgressBar::setPrintMultiline(bool printMultiline)
-{
-    this->printMultiline = printMultiline;
-}
+    void setPrintMultiline(bool printMultiline) override
+    {
+        this->printMultiline = printMultiline;
+    }
+};
 
 Logger * makeProgressBar()
 {

src/libmain/progress-bar.hh

@@ -1,135 +1,10 @@
 #pragma once
 ///@file
 
-#include <chrono>
-
 #include "logging.hh"
-#include "sync.hh"
 
 namespace nix {
 
-// 100 years ought to be enough for anyone (yet sufficiently smaller than max() to not cause signed integer overflow).
-constexpr const auto A_LONG_TIME = std::chrono::duration_cast<std::chrono::milliseconds>(
-    100 * 365 * std::chrono::seconds(86400)
-);
-
-struct ProgressBar : public Logger
-{
-    struct ActInfo
-    {
-        using TimePoint = std::chrono::time_point<std::chrono::steady_clock>;
-
-        std::string s, lastLine, phase;
-        ActivityType type = actUnknown;
-        uint64_t done = 0;
-        uint64_t expected = 0;
-        uint64_t running = 0;
-        uint64_t failed = 0;
-        std::map<ActivityType, uint64_t> expectedByType;
-        bool visible = true;
-        ActivityId parent;
-        std::optional<std::string> name;
-        TimePoint startTime;
-    };
-
-    struct ActivitiesByType
-    {
-        std::map<ActivityId, std::list<ActInfo>::iterator> its;
-        uint64_t done = 0;
-        uint64_t expected = 0;
-        uint64_t failed = 0;
-    };
-
-    struct State
-    {
-        std::list<ActInfo> activities;
-        std::map<ActivityId, std::list<ActInfo>::iterator> its;
-
-        std::map<ActivityType, ActivitiesByType> activitiesByType;
-
-        int lastLines = 0;
-
-        uint64_t filesLinked = 0, bytesLinked = 0;
-
-        uint64_t corruptedPaths = 0, untrustedPaths = 0;
-
-        bool active = true;
-        bool paused = false;
-        bool haveUpdate = true;
-    };
-
-    Sync<State> state_;
-
-    std::thread updateThread;
-
-    std::condition_variable quitCV, updateCV;
-
-    bool printBuildLogs = false;
-    bool printMultiline = false;
-    bool isTTY;
-
-    ProgressBar(bool isTTY)
-        : isTTY(isTTY)
-    {
-        state_.lock()->active = isTTY;
-        updateThread = std::thread([&]() {
-            auto state(state_.lock());
-            auto nextWakeup = A_LONG_TIME;
-            while (state->active) {
-                if (!state->haveUpdate)
-                    state.wait_for(updateCV, nextWakeup);
-                nextWakeup = draw(*state, {});
-                state.wait_for(quitCV, std::chrono::milliseconds(50));
-            }
-        });
-    }
-
-    ~ProgressBar();
-
-    void stop() override final;
-
-    void pause() override;
-
-    void resume() override;
-
-    bool isVerbose() override;
-
-    void log(Verbosity lvl, std::string_view s) override;
-
-    void logEI(const ErrorInfo & ei) override;
-
-    void log(State & state, Verbosity lvl, std::string_view s);
-
-    void startActivity(
-        ActivityId act,
-        Verbosity lvl,
-        ActivityType type,
-        const std::string & s,
-        const Fields & fields,
-        ActivityId parent
-    ) override;
-
-    bool hasAncestor(State & state, ActivityType type, ActivityId act);
-
-    void stopActivity(ActivityId act) override;
-
-    void result(ActivityId act, ResultType type, const std::vector<Field> & fields) override;
-
-    void update(State & state);
-
-    std::chrono::milliseconds draw(State & state, const std::optional<std::string_view> & s);
-
-    std::string getStatus(State & state);
-
-    void writeToStdout(std::string_view s) override;
-
-    std::optional<char> ask(std::string_view msg) override;
-
-    void setPrintBuildLogs(bool printBuildLogs) override;
-
-    void setPrintMultiline(bool printMultiline) override;
-};
-
 Logger * makeProgressBar();
 
 void startProgressBar();

src/libstore/binary-cache-store.cc

@@ -38,7 +38,7 @@ void BinaryCacheStore::init()
 {
     std::string cacheInfoFile = "nix-cache-info";
 
-    auto cacheInfo = getFileContents(cacheInfoFile);
+    auto cacheInfo = getFile(cacheInfoFile);
     if (!cacheInfo) {
         upsertFile(cacheInfoFile, "StoreDir: " + storeDir + "\n", "text/x-nix-cache-info");
     } else {
@@ -69,10 +69,10 @@ void BinaryCacheStore::upsertFile(const std::string & path,
 
 void BinaryCacheStore::getFile(const std::string & path, Sink & sink)
 {
-    sink(*getFileContents(path));
+    sink(*getFile(path));
 }
 
-std::optional<std::string> BinaryCacheStore::getFileContents(const std::string & path)
+std::optional<std::string> BinaryCacheStore::getFile(const std::string & path)
 {
     StringSink sink;
     try {
@@ -359,7 +359,7 @@ std::shared_ptr<const ValidPathInfo> BinaryCacheStore::queryPathInfoUncached(con
 
     auto narInfoFile = narInfoFileFor(storePath);
 
-    auto data = getFileContents(narInfoFile);
+    auto data = getFile(narInfoFile);
 
     if (!data) return nullptr;
 
@@ -385,7 +385,7 @@ StorePath BinaryCacheStore::addToStore(
     if (method == FileIngestionMethod::Recursive) {
         dumpPath(srcPath, sink, filter);
     } else {
-        readFileSource(srcPath)->drainInto(sink);
+        readFile(srcPath, sink);
     }
     auto h = sink.finish().first;
 
@@ -446,7 +446,7 @@ std::shared_ptr<const Realisation> BinaryCacheStore::queryRealisationUncached(co
 {
     auto outputInfoFilePath = realisationsPrefix + "/" + id.to_string() + ".doi";
 
-    auto data = getFileContents(outputInfoFilePath);
+    auto data = getFile(outputInfoFilePath);
     if (!data) return {};
 
     auto realisation = Realisation::fromJSON(
@@ -486,7 +486,7 @@ std::optional<std::string> BinaryCacheStore::getBuildLogExact(const StorePath &
 
     debug("fetching build log from binary cache '%s/%s'", getUri(), logPath);
 
-    return getFileContents(logPath);
+    return getFile(logPath);
 }
 
 void BinaryCacheStore::addBuildLog(const StorePath & drvPath, std::string_view log)

src/libstore/binary-cache-store.hh

@@ -85,7 +85,7 @@ public:
      */
     virtual void getFile(const std::string & path, Sink & sink);
 
-    virtual std::optional<std::string> getFileContents(const std::string & path);
+    virtual std::optional<std::string> getFile(const std::string & path);
 
 public:
 

src/libstore/build/local-derivation-goal.cc

@@ -2981,7 +2981,7 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
             HashModuloSink caSink { outputHash.hashType, oldHashPart };
             std::visit(overloaded {
                 [&](const TextIngestionMethod &) {
-                    readFileSource(actualPath)->drainInto(caSink);
+                    readFile(actualPath, caSink);
                 },
                 [&](const FileIngestionMethod & m2) {
                     switch (m2) {
@@ -2989,7 +2989,7 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
                         dumpPath(actualPath, caSink);
                         break;
                     case FileIngestionMethod::Flat:
-                        readFileSource(actualPath)->drainInto(caSink);
+                        readFile(actualPath, caSink);
                         break;
                     }
                 },

src/libstore/builtins/fetchurl.cc

@@ -41,7 +41,7 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData)
 
             auto decompressor = makeDecompressionSink(
                 unpack && mainUrl.ends_with(".xz") ? "xz" : "none", sink);
-            fileTransfer->download(std::move(request))->drainInto(*decompressor);
+            fileTransfer->download(std::move(request), *decompressor);
             decompressor->finish();
         });
 

src/libstore/daemon.cc

@@ -296,6 +296,17 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
+    case WorkerProto::Op::HasSubstitutes: {
+        auto path = store->parseStorePath(readString(from));
+        logger->startWork();
+        StorePathSet paths; // FIXME
+        paths.insert(path);
+        auto res = store->querySubstitutablePaths(paths);
+        logger->stopWork();
+        to << (res.count(path) != 0);
+        break;
+    }
+
     case WorkerProto::Op::QuerySubstitutablePaths: {
         auto paths = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
         logger->startWork();
@@ -305,74 +316,36 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
-    case WorkerProto::Op::HasSubstitutes: {
-        throw UnimplementedError("HasSubstitutes is not supported in Lix. This is not used if the declared server protocol is > 1.12 (Nix 1.0, 2012)");
-        break;
-    }
-
     case WorkerProto::Op::QueryPathHash: {
-        throw UnimplementedError("QueryPathHash is not supported in Lix, client usages were removed in 2016 in e0204f8d462041387651af388074491fd0bf36d6");
-        break;
-    }
-
-    case WorkerProto::Op::QueryReferences: {
-        throw UnimplementedError("QueryReferences is not supported in Lix, client usages were removed in 2016 in e0204f8d462041387651af388074491fd0bf36d6");
-        break;
-    }
-
-    case WorkerProto::Op::QueryDeriver: {
-        throw UnimplementedError("QueryDeriver is not supported in Lix, client usages were removed in 2016 in e0204f8d462041387651af388074491fd0bf36d6");
-        break;
-    }
-
-    case WorkerProto::Op::ExportPath: {
-        throw UnimplementedError("ExportPath is not supported in Lix, client usage were removed in 2017 in 27dc76c1a5dbe654465245ff5f6bc22e2c8902da");
-        break;
-    }
-
-    case WorkerProto::Op::ImportPaths: {
-        throw UnimplementedError("ImportPaths is not supported in Lix. This is not used if the declared server protocol is >= 1.18 (Nix 2.0, 2016)");
+        auto path = store->parseStorePath(readString(from));
+        logger->startWork();
+        auto hash = store->queryPathInfo(path)->narHash;
+        logger->stopWork();
+        to << hash.to_string(Base16, false);
         break;
     }
 
+    case WorkerProto::Op::QueryReferences:
     case WorkerProto::Op::QueryReferrers:
     case WorkerProto::Op::QueryValidDerivers:
     case WorkerProto::Op::QueryDerivationOutputs: {
         auto path = store->parseStorePath(readString(from));
         logger->startWork();
         StorePathSet paths;
-
-        #pragma GCC diagnostic push
-        #pragma GCC diagnostic ignored "-Wswitch-enum"
-        switch (op) {
-            case WorkerProto::Op::QueryReferrers: {
-                store->queryReferrers(path, paths);
-                break;
-            }
-            case WorkerProto::Op::QueryValidDerivers: {
-                paths = store->queryValidDerivers(path);
-                break;
-            }
-            case WorkerProto::Op::QueryDerivationOutputs: {
-                // Only sent if server presents proto version <= 1.21
-                REMOVE_AFTER_DROPPING_PROTO_MINOR(21);
-                paths = store->queryDerivationOutputs(path);
-                break;
-            }
-            default:
-                abort();
-                break;
-        }
-        #pragma GCC diagnostic pop
-
+        if (op == WorkerProto::Op::QueryReferences)
+            for (auto & i : store->queryPathInfo(path)->references)
+                paths.insert(i);
+        else if (op == WorkerProto::Op::QueryReferrers)
+            store->queryReferrers(path, paths);
+        else if (op == WorkerProto::Op::QueryValidDerivers)
+            paths = store->queryValidDerivers(path);
+        else paths = store->queryDerivationOutputs(path);
         logger->stopWork();
         WorkerProto::write(*store, wconn, paths);
         break;
     }
 
     case WorkerProto::Op::QueryDerivationOutputNames: {
-        // Unused in CppNix >= 2.4 (removed in 045b07200c77bf1fe19c0a986aafb531e7e1ba54)
-        REMOVE_AFTER_DROPPING_PROTO_MINOR(31);
         auto path = store->parseStorePath(readString(from));
         logger->startWork();
         auto names = store->readDerivation(path).outputNames();
@@ -390,6 +363,15 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
+    case WorkerProto::Op::QueryDeriver: {
+        auto path = store->parseStorePath(readString(from));
+        logger->startWork();
+        auto info = store->queryPathInfo(path);
+        logger->stopWork();
+        to << (info->deriver ? store->printStorePath(*info->deriver) : "");
+        break;
+    }
+
     case WorkerProto::Op::QueryPathFromHashPart: {
         auto hashPart = readString(from);
         logger->startWork();
@@ -511,6 +493,29 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
+    case WorkerProto::Op::ExportPath: {
+        auto path = store->parseStorePath(readString(from));
+        readInt(from); // obsolete
+        logger->startWork();
+        TunnelSink sink(to);
+        store->exportPath(path, sink);
+        logger->stopWork();
+        to << 1;
+        break;
+    }
+
+    case WorkerProto::Op::ImportPaths: {
+        logger->startWork();
+        TunnelSource source(from, to);
+        auto paths = store->importPaths(source,
+            trusted ? NoCheckSigs : CheckSigs);
+        logger->stopWork();
+        Strings paths2;
+        for (auto & i : paths) paths2.push_back(store->printStorePath(i));
+        to << paths2;
+        break;
+    }
+
     case WorkerProto::Op::BuildPaths: {
         auto drvs = WorkerProto::Serialise<DerivedPaths>::read(*store, rconn);
         BuildMode mode = buildModeFromInteger(readInt(from));
@@ -661,10 +666,8 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
-    // Obsolete since 9947f1646a26b339fff2e02b77798e9841fac7f0 (included in CppNix 2.5.0).
+    // Obsolete.
     case WorkerProto::Op::SyncWithGC: {
-        // CppNix 2.5.0 is 32
-        REMOVE_AFTER_DROPPING_PROTO_MINOR(31);
         logger->startWork();
         logger->stopWork();
         to << 1;

src/libstore/filetransfer.cc

@@ -686,8 +686,16 @@ struct curlFileTransfer : public FileTransfer
             ->callback.get_future();
     }
 
-    box_ptr<Source> download(FileTransferRequest && request) override
+    void download(FileTransferRequest && request, Sink & sink) override
     {
+        /* Note: we can't call 'sink' via request.dataCallback, because
+           that would cause the sink to execute on the fileTransfer
+           thread. If 'sink' is a coroutine, this will fail. Also, if the
+           sink is expensive (e.g. one that does decompression and writing
+           to the Nix store), it would stall the download thread too much.
+           Therefore we use a buffer to communicate data between the
+           download thread and the calling thread. */
+
         struct State {
             bool done = false, failed = false;
             std::exception_ptr exc;
@@ -697,6 +705,13 @@ struct curlFileTransfer : public FileTransfer
 
         auto _state = std::make_shared<Sync<State>>();
 
+        /* In case of an exception, wake up the download thread. */
+        Finally finally([&]() {
+            auto state(_state->lock());
+            state->failed |= std::uncaught_exceptions() != 0;
+            state->request.notify_one();
+        });
+
         enqueueFileTransfer(
             request,
             [_state](std::exception_ptr ex) {
@@ -735,99 +750,50 @@ struct curlFileTransfer : public FileTransfer
             }
         );
 
-        struct InnerSource : Source
-        {
-            const std::shared_ptr<Sync<State>> _state;
+        std::unique_ptr<FinishSink> decompressor;
+
+        while (true) {
+            checkInterrupt();
+
             std::string chunk;
-            std::string_view buffered;
 
-            explicit InnerSource(const std::shared_ptr<Sync<State>> & state) : _state(state) {}
-
-            ~InnerSource()
+            /* Grab data if available, otherwise wait for the download
+               thread to wake us up. */
             {
-                // wake up the download thread if it's still going and have it abort
                 auto state(_state->lock());
-                state->failed |= !state->done;
+
+                if (state->data.empty()) {
+
+                    if (state->done) {
+                        if (state->exc) std::rethrow_exception(state->exc);
+                        if (decompressor) {
+                            decompressor->finish();
+                        }
+                        return;
+                    }
+
+                    state.wait(state->avail);
+
+                    if (state->data.empty()) continue;
+                }
+
+                chunk = std::move(state->data);
+                /* Reset state->data after the move, since we check data.empty() */
+                state->data = "";
+
+                if (!decompressor) {
+                    decompressor = makeDecompressionSink(state->encoding, sink);
+                }
+
                 state->request.notify_one();
             }
 
-            void awaitData(Sync<State>::Lock & state)
-            {
-                /* Grab data if available, otherwise wait for the download
-                   thread to wake us up. */
-                while (buffered.empty()) {
-                    if (state->data.empty()) {
-                        if (state->done) {
-                            if (state->exc) {
-                                std::rethrow_exception(state->exc);
-                            }
-                            return;
-                        }
-
-                        state.wait(state->avail);
-                    }
-
-                    chunk = std::move(state->data);
-                    buffered = chunk;
-                    state->request.notify_one();
-                }
-            }
-
-            size_t read(char * data, size_t len) override
-            {
-                auto readPartial = [this](char * data, size_t len) {
-                    const auto available = std::min(len, buffered.size());
-                    memcpy(data, buffered.data(), available);
-                    buffered.remove_prefix(available);
-                    return available;
-                };
-                size_t total = readPartial(data, len);
-
-                while (total < len) {
-                    {
-                        auto state(_state->lock());
-                        awaitData(state);
-                    }
-                    const auto current = readPartial(data + total, len - total);
-                    total += current;
-                    if (total == 0 || current == 0) {
-                        break;
-                    }
-                }
-
-                if (total == 0) {
-                    throw EndOfFile("download finished");
-                }
-
-                return total;
-            }
-        };
-
-        struct DownloadSource : Source
-        {
-            InnerSource inner;
-            std::unique_ptr<Source> decompressor;
-
-            explicit DownloadSource(const std::shared_ptr<Sync<State>> & state) : inner(state) {}
-
-            size_t read(char * data, size_t len) override
-            {
-                checkInterrupt();
-
-                if (!decompressor) {
-                    auto state(inner._state->lock());
-                    inner.awaitData(state);
-                    decompressor = makeDecompressionSource(state->encoding, inner);
-                }
-
-                return decompressor->read(data, len);
-            }
-        };
-
-        auto source = make_box_ptr<DownloadSource>(_state);
-        auto lock(_state->lock());
-        source->inner.awaitData(lock);
-        return source;
+            /* Flush the data to the sink and wake up the download thread
+               if it's blocked on a full buffer. We don't hold the state
+               lock while doing this to prevent blocking the download
+               thread if sink() takes a long time. */
+            (*decompressor)(chunk);
+        }
     }
 };
 

src/libstore/filetransfer.hh

@@ -1,7 +1,6 @@
 #pragma once
 ///@file
 
-#include "box_ptr.hh"
 #include "logging.hh"
 #include "serialise.hh"
 #include "types.hh"
@@ -105,13 +104,10 @@ struct FileTransfer
     FileTransferResult transfer(const FileTransferRequest & request);
 
     /**
-     * Download a file, returning its contents through a source. Will not return
-     * before the transfer has fully started, ensuring that any errors thrown by
-     * the setup phase (e.g. HTTP 404 or similar errors) are not postponed to be
-     * thrown by the returned source. The source will only throw errors detected
-     * during the transfer itself (decompression errors, connection drops, etc).
+     * Download a file, writing its data to a sink. The sink will be
+     * invoked on the thread of the caller.
      */
-    virtual box_ptr<Source> download(FileTransferRequest && request) = 0;
+    virtual void download(FileTransferRequest && request, Sink & sink) = 0;
 
     enum Error { NotFound, Forbidden, Misc, Transient, Interrupted };
 };

src/libstore/http-binary-cache-store.cc

@@ -155,7 +155,7 @@ protected:
         checkEnabled();
         auto request(makeRequest(path));
         try {
-            getFileTransfer()->download(std::move(request))->drainInto(sink);
+            getFileTransfer()->download(std::move(request), sink);
         } catch (FileTransferError & e) {
             if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden)
                 throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri());
@@ -164,7 +164,7 @@ protected:
         }
     }
 
-    std::optional<std::string> getFileContents(const std::string & path) override
+    std::optional<std::string> getFile(const std::string & path) override
     {
         checkEnabled();
 

src/libstore/local-binary-cache-store.cc

@@ -71,7 +71,7 @@ protected:
     void getFile(const std::string & path, Sink & sink) override
     {
         try {
-            readFileSource(binaryCacheDir + "/" + path)->drainInto(sink);
+            readFile(binaryCacheDir + "/" + path, sink);
         } catch (SysError & e) {
             if (e.errNo == ENOENT)
                 throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache", path);

src/libstore/local-store.cc

@@ -1890,7 +1890,7 @@ ContentAddress LocalStore::hashCAPath(
     HashModuloSink caSink ( hashType, std::string(pathHash) );
     std::visit(overloaded {
         [&](const TextIngestionMethod &) {
-            readFileSource(path)->drainInto(caSink);
+            readFile(path, caSink);
         },
         [&](const FileIngestionMethod & m2) {
             switch (m2) {
@@ -1898,7 +1898,7 @@ ContentAddress LocalStore::hashCAPath(
                 dumpPath(path, caSink);
                 break;
             case FileIngestionMethod::Flat:
-                readFileSource(path)->drainInto(caSink);
+                readFile(path, caSink);
                 break;
             }
         },

src/libstore/remote-store.cc

@@ -307,7 +307,6 @@ StorePathSet RemoteStore::queryDerivationOutputs(const StorePath & path)
     if (GET_PROTOCOL_MINOR(getProtocol()) >= 22) {
         return Store::queryDerivationOutputs(path);
     }
-    REMOVE_AFTER_DROPPING_PROTO_MINOR(21);
     auto conn(getConnection());
     conn->to << WorkerProto::Op::QueryDerivationOutputs << printStorePath(path);
     conn.processStderr();
@@ -337,7 +336,6 @@ std::map<std::string, std::optional<StorePath>> RemoteStore::queryPartialDerivat
             return outputs;
         }
     } else {
-        REMOVE_AFTER_DROPPING_PROTO_MINOR(21);
         auto & evalStore = evalStore_ ? *evalStore_ : *this;
         // Fallback for old daemon versions.
         // For floating-CA derivations (and their co-dependencies) this is an
@@ -532,7 +530,6 @@ void RemoteStore::registerDrvOutput(const Realisation & info)
     auto conn(getConnection());
     conn->to << WorkerProto::Op::RegisterDrvOutput;
     if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
-        REMOVE_AFTER_DROPPING_PROTO_MINOR(30);
         conn->to << info.id.to_string();
         conn->to << std::string(info.outPath.to_string());
     } else {
@@ -620,7 +617,6 @@ std::vector<KeyedBuildResult> RemoteStore::buildPathsWithResults(
         conn.processStderr();
         return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*this, *conn);
     } else {
-        REMOVE_AFTER_DROPPING_PROTO_MINOR(33);
         // Avoid deadlock.
         conn_.reset();
 

src/libstore/serve-protocol.hh

@@ -8,11 +8,6 @@ namespace nix {
 #define SERVE_MAGIC_1 0x390c9deb
 #define SERVE_MAGIC_2 0x5452eecb
 
-// This must remain at 2.7 (Nix 2.18) forever in Lix, since the protocol
-// versioning is monotonic, so if we ever change it in the future, it will
-// break compatibility with any potential CppNix-originated protocol changes.
-//
-// Lix intends to replace this protocol entirely.
 #define SERVE_PROTOCOL_VERSION (2 << 8 | 7)
 #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)

src/libstore/store-api.cc

@@ -279,7 +279,7 @@ StorePath Store::addToStore(
         if (method == FileIngestionMethod::Recursive)
             dumpPath(srcPath, sink, filter);
         else
-            readFileSource(srcPath)->drainInto(sink);
+            readFile(srcPath, sink);
     });
     return addToStoreFromDump(*source, name, method, hashAlgo, repair, references);
 }
@@ -803,31 +803,17 @@ StorePathSet Store::queryValidPaths(const StorePathSet & paths, SubstituteFlag m
 
     auto doQuery = [&](const StorePath & path) {
         checkInterrupt();
-
-        bool exists = false;
-        std::exception_ptr newExc{};
-
+        auto state(state_.lock());
         try {
-            queryPathInfo(path);
-            exists = true;
+            auto info = queryPathInfo(path);
+            state->valid.insert(path);
         } catch (InvalidPath &) {
         } catch (...) {
-            newExc = std::current_exception();
-        }
-
-        {
-            auto state(state_.lock());
-
-            if (exists) {
-                state->valid.insert(path);
-            }
-            if (newExc != nullptr) {
-                state->exc = newExc;
-            }
-            assert(state->left);
-            if (!--state->left)
-                wakeup.notify_one();
+            state->exc = std::current_exception();
         }
+        assert(state->left);
+        if (!--state->left)
+            wakeup.notify_one();
     };
 
     for (auto & path : paths)

src/libstore/worker-protocol.hh

@@ -23,10 +23,6 @@ namespace nix {
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)
 
 
-#define REMOVE_AFTER_DROPPING_PROTO_MINOR(protoMinor) \
-    static_assert(MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION <= (protoMinor))
-
-
 #define STDERR_NEXT  0x6f6c6d67
 #define STDERR_READ  0x64617461 // data needed from source
 #define STDERR_WRITE 0x64617416 // data for sink
@@ -140,30 +136,30 @@ struct WorkerProto
 enum struct WorkerProto::Op : uint64_t
 {
     IsValidPath = 1,
-    HasSubstitutes = 3, // obsolete since 2012, stubbed to error
-    QueryPathHash = 4, // obsolete since 2016, stubbed to error
-    QueryReferences = 5, // obsolete since 2016, stubbed to error
+    HasSubstitutes = 3,
+    QueryPathHash = 4, // obsolete
+    QueryReferences = 5, // obsolete
     QueryReferrers = 6,
     AddToStore = 7,
-    AddTextToStore = 8, // obsolete since protocol 1.25, CppNix 2.4. Use WorkerProto::Op::AddToStore
+    AddTextToStore = 8, // obsolete since 1.25, Nix 3.0. Use WorkerProto::Op::AddToStore
     BuildPaths = 9,
     EnsurePath = 10,
     AddTempRoot = 11,
     AddIndirectRoot = 12,
-    SyncWithGC = 13, // obsolete since CppNix 2.5.0
+    SyncWithGC = 13,
     FindRoots = 14,
-    ExportPath = 16, // obsolete since 2017, stubbed to error
-    QueryDeriver = 18, // obsolete since 2016, stubbed to error
+    ExportPath = 16, // obsolete
+    QueryDeriver = 18, // obsolete
     SetOptions = 19,
     CollectGarbage = 20,
     QuerySubstitutablePathInfo = 21,
-    QueryDerivationOutputs = 22, // obsolete since protocol 1.21, CppNix 2.4
+    QueryDerivationOutputs = 22, // obsolete
     QueryAllValidPaths = 23,
-    QueryFailedPaths = 24, // obsolete, removed
-    ClearFailedPaths = 25, // obsolete, removed
+    QueryFailedPaths = 24,
+    ClearFailedPaths = 25,
     QueryPathInfo = 26,
-    ImportPaths = 27, // obsolete since 2016
-    QueryDerivationOutputNames = 28, // obsolete since CppNix 2.4
+    ImportPaths = 27, // obsolete
+    QueryDerivationOutputNames = 28, // obsolete
     QueryPathFromHashPart = 29,
     QuerySubstitutablePathInfos = 30,
     QueryValidPaths = 31,

src/libutil/compression.cc

@@ -163,24 +163,23 @@ struct BrotliDecompressionSource : Source
         uint8_t * out = (uint8_t *) data;
         const auto * begin = out;
 
-        while (len && !BrotliDecoderIsFinished(state.get())) {
-            checkInterrupt();
+        try {
+            while (len && !BrotliDecoderIsFinished(state.get())) {
+                checkInterrupt();
 
-            while (avail_in == 0) {
-                try {
+                while (avail_in == 0) {
                     avail_in = inner->read(buf.get(), BUF_SIZE);
-                } catch (EndOfFile &) {
-                    break;
+                    next_in = (const uint8_t *) buf.get();
                 }
-                next_in = (const uint8_t *) buf.get();
-            }
 
-            if (!BrotliDecoderDecompressStream(
-                    state.get(), &avail_in, &next_in, &len, &out, nullptr
-                ))
-            {
-                throw CompressionError("error while decompressing brotli file");
+                if (!BrotliDecoderDecompressStream(
+                        state.get(), &avail_in, &next_in, &len, &out, nullptr
+                    ))
+                {
+                    throw CompressionError("error while decompressing brotli file");
+                }
             }
+        } catch (EndOfFile &) {
         }
 
         if (begin != out) {
@@ -193,9 +192,11 @@ struct BrotliDecompressionSource : Source
 
 std::string decompress(const std::string & method, std::string_view in)
 {
-    StringSource src{in};
-    auto filter = makeDecompressionSource(method, src);
-    return filter->drain();
+    StringSink ssink;
+    auto sink = makeDecompressionSink(method, ssink);
+    (*sink)(in);
+    sink->finish();
+    return std::move(ssink.s);
 }
 
 std::unique_ptr<FinishSink> makeDecompressionSink(const std::string & method, Sink & nextSink)
@@ -223,19 +224,6 @@ std::unique_ptr<FinishSink> makeDecompressionSink(const std::string & method, Si
         });
 }
 
-std::unique_ptr<Source> makeDecompressionSource(const std::string & method, Source & inner)
-{
-    if (method == "none" || method == "") {
-        return std::make_unique<LambdaSource>([&](char * data, size_t len) {
-            return inner.read(data, len);
-        });
-    } else if (method == "br") {
-        return std::make_unique<BrotliDecompressionSource>(inner);
-    } else {
-        return std::make_unique<ArchiveDecompressionSource>(inner);
-    }
-}
-
 struct BrotliCompressionSink : ChunkedCompressionSink
 {
     Sink & nextSink;

src/libutil/compression.hh

@@ -19,7 +19,6 @@ struct CompressionSink : BufferedSink, FinishSink
 std::string decompress(const std::string & method, std::string_view in);
 
 std::unique_ptr<FinishSink> makeDecompressionSink(const std::string & method, Sink & nextSink);
-std::unique_ptr<Source> makeDecompressionSource(const std::string & method, Source & inner);
 
 std::string compress(const std::string & method, std::string_view in, const bool parallel = false, int level = -1);
 

src/libutil/file-system.cc

@@ -289,17 +289,12 @@ std::string readFile(const Path & path)
 }
 
 
-box_ptr<Source> readFileSource(const Path & path)
+void readFile(const Path & path, Sink & sink)
 {
     AutoCloseFD fd{open(path.c_str(), O_RDONLY | O_CLOEXEC)};
     if (!fd)
         throw SysError("opening file '%s'", path);
-
-    struct FileSource : FdSource {
-        AutoCloseFD fd;
-        explicit FileSource(AutoCloseFD fd) : FdSource(fd.get()), fd(std::move(fd)) {}
-    };
-    return make_box_ptr<FileSource>(std::move(fd));
+    drainFD(fd.get(), sink);
 }
 
 

src/libutil/file-system.hh

@@ -5,7 +5,6 @@
  * Utiltities for working with the file sytem and file paths.
  */
 
-#include "box_ptr.hh"
 #include "types.hh"
 #include "file-descriptor.hh"
 
@@ -143,7 +142,7 @@ unsigned char getFileType(const Path & path);
  * Read the contents of a file into a string.
  */
 std::string readFile(const Path & path);
-box_ptr<Source> readFileSource(const Path & path);
+void readFile(const Path & path, Sink & sink);
 
 /**
  * Write a string to a file.

src/libutil/hash.cc

@@ -324,7 +324,7 @@ Hash hashString(HashType ht, std::string_view s)
 Hash hashFile(HashType ht, const Path & path)
 {
     HashSink sink(ht);
-    readFileSource(path)->drainInto(sink);
+    readFile(path, sink);
     return sink.finish().first;
 }
 

src/nix/add-to-store.cc

@@ -37,7 +37,7 @@ struct CmdAddToStore : MixDryRun, StoreCommand
         Hash hash = narHash;
         if (ingestionMethod == FileIngestionMethod::Flat) {
             HashSink hsink(htSHA256);
-            readFileSource(path)->drainInto(hsink);
+            readFile(path, hsink);
             hash = hsink.finish().first;
         }
 

src/nix/bundle.cc

@@ -1,5 +1,5 @@
 #include "installable-flake.hh"
-#include "command.hh"
+#include "command-installable-value.hh"
 #include "common-args.hh"
 #include "shared.hh"
 #include "store-api.hh"
@@ -9,7 +9,7 @@
 
 using namespace nix;
 
-struct CmdBundle : InstallableCommand
+struct CmdBundle : InstallableValueCommand
 {
     std::string bundler = "github:NixOS/bundlers";
     std::optional<Path> outLink;
@@ -71,13 +71,11 @@ struct CmdBundle : InstallableCommand
         return res;
     }
 
-    void run(ref<Store> store, ref<Installable> installable) override
+    void run(ref<Store> store, ref<InstallableValue> installable) override
     {
         auto evalState = getEvalState();
 
-        auto const installableValue = InstallableValue::require(installable);
-
-        auto val = installableValue->toValue(*evalState).first;
+        auto val = installable->toValue(*evalState).first;
 
         auto [bundlerFlakeRef, bundlerName, extendedOutputsSpec] = parseFlakeRefWithFragmentAndExtendedOutputsSpec(bundler, absPath("."));
         const flake::LockFlags lockFlags{ .writeLockFile = false };

src/nix/develop.cc

@@ -1,6 +1,6 @@
 #include "eval.hh"
 #include "installable-flake.hh"
-#include "command.hh"
+#include "command-installable-value.hh"
 #include "common-args.hh"
 #include "shared.hh"
 #include "store-api.hh"

src/nix/edit.cc

@@ -1,4 +1,4 @@
-#include "command.hh"
+#include "command-installable-value.hh"
 #include "shared.hh"
 #include "eval.hh"
 #include "attr-path.hh"
@@ -10,7 +10,7 @@
 
 using namespace nix;
 
-struct CmdEdit : InstallableCommand
+struct CmdEdit : InstallableValueCommand
 {
     std::string description() override
     {
@@ -26,19 +26,17 @@ struct CmdEdit : InstallableCommand
 
     Category category() override { return catSecondary; }
 
-    void run(ref<Store> store, ref<Installable> installable) override
+    void run(ref<Store> store, ref<InstallableValue> installable) override
     {
         auto state = getEvalState();
 
-        auto const installableValue = InstallableValue::require(installable);
-
         const auto [file, line] = [&] {
-            auto [v, pos] = installableValue->toValue(*state);
+            auto [v, pos] = installable->toValue(*state);
 
             try {
                 return findPackageFilename(*state, *v, installable->what());
             } catch (NoPositionInfo &) {
-                throw Error("cannot find position information for '%s", installableValue->what());
+                throw Error("cannot find position information for '%s", installable->what());
             }
         }();
 

src/nix/eval.cc

@@ -1,4 +1,4 @@
-#include "command.hh"
+#include "command-installable-value.hh"
 #include "common-args.hh"
 #include "print-options.hh"
 #include "shared.hh"
@@ -12,13 +12,13 @@
 
 using namespace nix;
 
-struct CmdEval : MixJSON, InstallableCommand, MixReadOnlyOption
+struct CmdEval : MixJSON, InstallableValueCommand, MixReadOnlyOption
 {
     bool raw = false;
     std::optional<std::string> apply;
     std::optional<Path> writeTo;
 
-    CmdEval() : InstallableCommand()
+    CmdEval() : InstallableValueCommand()
     {
         addFlag({
             .longName = "raw",
@@ -55,16 +55,14 @@ struct CmdEval : MixJSON, InstallableCommand, MixReadOnlyOption
 
     Category category() override { return catSecondary; }
 
-    void run(ref<Store> store, ref<Installable> installable) override
+    void run(ref<Store> store, ref<InstallableValue> installable) override
     {
         if (raw && json)
             throw UsageError("--raw and --json are mutually exclusive");
 
-        auto const installableValue = InstallableValue::require(installable);
-
         auto state = getEvalState();
 
-        auto [v, pos] = installableValue->toValue(*state);
+        auto [v, pos] = installable->toValue(*state);
         NixStringContext context;
 
         if (apply) {

src/nix/hash.cc

@@ -85,7 +85,7 @@ struct CmdHashBase : Command
 
             switch (mode) {
             case FileIngestionMethod::Flat:
-                readFileSource(path)->drainInto(*hashSink);
+                readFile(path, *hashSink);
                 break;
             case FileIngestionMethod::Recursive:
                 dumpPath(path, *hashSink);

src/nix/prefetch.cc

@@ -98,7 +98,7 @@ std::tuple<StorePath, Hash> prefetchFile(
             FdSink sink(fd.get());
 
             FileTransferRequest req(url);
-            getFileTransfer()->download(std::move(req))->drainInto(sink);
+            getFileTransfer()->download(std::move(req), sink);
         }
 
         /* Optionally unpack the file. */

src/nix/run.cc

@@ -1,7 +1,6 @@
 #include "run.hh"
-#include "command.hh"
+#include "command-installable-value.hh"
 #include "common-args.hh"
-#include "installables.hh"
 #include "shared.hh"
 #include "store-api.hh"
 #include "derivations.hh"
@@ -146,7 +145,7 @@ struct CmdShell : InstallablesCommand, MixEnvironment
 
 static auto rCmdShell = registerCommand<CmdShell>("shell");
 
-struct CmdRun : InstallableCommand
+struct CmdRun : InstallableValueCommand
 {
     using InstallableCommand::run;
 
@@ -192,14 +191,12 @@ struct CmdRun : InstallableCommand
         return res;
     }
 
-    void run(ref<Store> store, ref<Installable> installable) override
+    void run(ref<Store> store, ref<InstallableValue> installable) override
     {
         auto state = getEvalState();
 
-        auto installableValue = InstallableValue::require(installable);
-
         lockFlags.applyNixConfig = true;
-        auto app = installableValue->toApp(*state).resolve(getEvalStore(), store);
+        auto app = installable->toApp(*state).resolve(getEvalStore(), store);
 
         Strings allArgs{app.program};
         for (auto & i : args) allArgs.push_back(i);

src/nix/search.cc

@@ -1,4 +1,4 @@
-#include "command.hh"
+#include "command-installable-value.hh"
 #include "globals.hh"
 #include "eval.hh"
 #include "eval-inline.hh"
@@ -23,7 +23,7 @@ std::string wrap(std::string prefix, std::string s)
     return concatStrings(prefix, s, ANSI_NORMAL);
 }
 
-struct CmdSearch : InstallableCommand, MixJSON
+struct CmdSearch : InstallableValueCommand, MixJSON
 {
     std::vector<std::string> res;
     std::vector<std::string> excludeRes;
@@ -62,10 +62,8 @@ struct CmdSearch : InstallableCommand, MixJSON
         };
     }
 
-    void run(ref<Store> store, ref<Installable> installable) override
+    void run(ref<Store> store, ref<InstallableValue> installable) override
     {
-        auto const installableValue = InstallableValue::require(installable);
-
         settings.readOnlyMode = true;
         evalSettings.enableImportFromDerivation.setDefault(false);
 
@@ -194,7 +192,7 @@ struct CmdSearch : InstallableCommand, MixJSON
             }
         };
 
-        for (auto & cursor : installableValue->getCursors(*state))
+        for (auto & cursor : installable->getCursors(*state))
             visit(*cursor, cursor->getAttrPath(), true);
 
         if (json)

tests/unit/libmain/progress-bar.cc

@@ -1,43 +0,0 @@
-#include <gtest/gtest.h>
-
-#include "eval.hh"
-#include "progress-bar.hh"
-#include "logging.hh"
-#include "shared.hh"
-
-constexpr std::string_view TEST_URL = "https://github.com/NixOS/nixpkgs/archive/master.tar.gz";
-// Arbitrary number. We picked the size of a Nixpkgs tarball that we downloaded.
-constexpr uint64_t TEST_EXPECTED = 43'370'307;
-// Arbitrary number. We picked the progress made on a Nixpkgs tarball download we interrupted.
-constexpr uint64_t TEST_DONE = 1'787'251;
-
-constexpr std::string_view EXPECTED = ANSI_GREEN "1.7" ANSI_NORMAL "/41.4 MiB DL";
-// Mostly here for informational purposes, but also if we change the way the escape codes
-// are defined this test might break in some annoying to debug way.
-constexpr std::string_view EXPECTED_RAW = "\x1b[32;1m1.7\x1b[0m/41.4 MiB DL";
-static_assert(EXPECTED == EXPECTED_RAW, "Hey, hey, the ANSI escape code definitions prolly changed");
-
-namespace nix
-{
-    TEST(ProgressBar, basicStatusRender) {
-        initNix();
-        initGC();
-
-        startProgressBar();
-        ASSERT_NE(dynamic_cast<ProgressBar *>(logger), nullptr);
-        ProgressBar & progressBar = dynamic_cast<ProgressBar &>(*logger);
-
-        Activity act(
-            progressBar,
-            lvlDebug,
-            actFileTransfer,
-            fmt("downloading '%s'", TEST_URL),
-            { "https://github.com/NixOS/nixpkgs/archive/master.tar.gz" }
-        );
-        act.progress(TEST_DONE, TEST_EXPECTED);
-        auto state = progressBar.state_.lock();
-        std::string const renderedStatus = progressBar.getStatus(*state);
-
-        ASSERT_EQ(renderedStatus, EXPECTED);
-    }
-}

tests/unit/libstore/filetransfer.cc

@@ -7,7 +7,6 @@
 #include <gtest/gtest.h>
 #include <netinet/in.h>
 #include <stdexcept>
-#include <string>
 #include <string_view>
 #include <sys/poll.h>
 #include <sys/socket.h>
@@ -137,7 +136,7 @@ TEST(FileTransfer, exceptionAbortsDownload)
 
     LambdaSink broken([](auto block) { throw Done(); });
 
-    ASSERT_THROW(ft->download(FileTransferRequest("file:///dev/zero"))->drainInto(broken), Done);
+    ASSERT_THROW(ft->download(FileTransferRequest("file:///dev/zero"), broken), Done);
 
     // makeFileTransfer returns a ref<>, which cannot be cleared. since we also
     // can't default-construct it we'll have to overwrite it instead, but we'll
@@ -160,21 +159,16 @@ TEST(FileTransfer, NOT_ON_DARWIN(reportsSetupErrors))
         FileTransferError);
 }
 
-TEST(FileTransfer, NOT_ON_DARWIN(defersFailures))
+TEST(FileTransfer, NOT_ON_DARWIN(reportsTransferError))
 {
-    auto [port, srv] = serveHTTP("200 ok", "content-length: 100000000\r\n", [] {
+    auto [port, srv] = serveHTTP("200 ok", "content-length: 100\r\n", [] {
         std::this_thread::sleep_for(10ms);
-        // just a bunch of data to fill the curl wrapper buffer, otherwise the
-        // initial wait for header data will also wait for the the response to
-        // complete (the source is only woken when curl returns data, and curl
-        // might only do so once its internal buffer has already been filled.)
-        return std::string(1024 * 1024, ' ');
+        return "";
     });
     auto ft = makeFileTransfer();
     FileTransferRequest req(fmt("http://[::1]:%d/index", port));
     req.baseRetryTimeMs = 0;
-    auto src = ft->download(std::move(req));
-    ASSERT_THROW(src->drain(), FileTransferError);
+    ASSERT_THROW(ft->transfer(req), FileTransferError);
 }
 
 TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
@@ -186,7 +180,7 @@ TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
     auto ft = makeFileTransfer();
 
     StringSink sink;
-    ft->download(FileTransferRequest(fmt("http://[::1]:%d/index", port)))->drainInto(sink);
+    ft->download(FileTransferRequest(fmt("http://[::1]:%d/index", port)), sink);
     EXPECT_EQ(sink.s, original);
 }
 

tests/unit/libutil/compression.cc

@@ -66,16 +66,6 @@ namespace nix {
         ASSERT_THROW(decompress(method, str), CompressionError);
     }
 
-    TEST(decompress, veryLongBrotli) {
-        auto method = "br";
-        auto str = std::string(65536, 'a');
-        auto o = decompress(method, compress(method, str));
-
-        // This is just to not print 64k of "a" for most failures
-        ASSERT_EQ(o.length(), str.length());
-        ASSERT_EQ(o, str);
-    }
-
     /* ----------------------------------------------------------------------------
      * compression sinks
      * --------------------------------------------------------------------------*/
@@ -91,17 +81,16 @@ namespace nix {
     }
 
     TEST(makeCompressionSink, compressAndDecompress) {
-        auto inputString = "slfja;sljfklsa;jfklsjfkl;sdjfkl;sadjfkl;sdjf;lsdfjsadlf";
-
         StringSink strSink;
-        auto sink = makeCompressionSink("bzip2", strSink);
+        auto inputString = "slfja;sljfklsa;jfklsjfkl;sdjfkl;sadjfkl;sdjf;lsdfjsadlf";
+        auto decompressionSink = makeDecompressionSink("bzip2", strSink);
+        auto sink = makeCompressionSink("bzip2", *decompressionSink);
+
         (*sink)(inputString);
         sink->finish();
+        decompressionSink->finish();
 
-        StringSource strSource{strSink.s};
-        auto decompressionSource = makeDecompressionSource("bzip2", strSource);
-
-        ASSERT_STREQ(decompressionSource->drain().c_str(), inputString);
+        ASSERT_STREQ(strSink.s.c_str(), inputString);
     }
 
 }

tests/unit/meson.build

@@ -249,25 +249,3 @@ test(
   suite : 'check',
   protocol : 'gtest',
 )
-
-libmain_tester = executable(
-  'liblixmain-tests',
-  files('libmain/progress-bar.cc'),
-  dependencies : [
-    liblixmain,
-    liblixexpr,
-    liblixutil,
-    liblixstore,
-    gtest,
-    boost,
-  ],
-  cpp_pch : cpp_pch,
-)
-
-test(
-  'libmain-unit-tests',
-  libmain_tester,
-  args : tests_args,
-  suite : 'check',
-  protocol : 'gtest',
-)