libstore/build: use an allowlist approach to syscall filtering

Previously, system call filtering (to prevent builders from storing files with
setuid/setgid permission bits or extended attributes) was performed using a
blocklist. While this looks simple at first, it actually carries significant
security and maintainability risks: after all, the kernel may add new syscalls
to achieve the same functionality one is trying to block, and it can even be
hard to actually add the syscall to the blocklist when building against a C
library that doesn't know about it yet. For a recent demonstration of this
happening in practice to Nix, see the introduction of fchmodat2 [0] [1].

The allowlist approach does not share the same drawback. While it does require
a rather large list of harmless syscalls to be maintained in the codebase,
failing to update this list (and roll out the update to all users) in time has
rather benign effects; at worst, very recent programs that already rely on new
syscalls will fail with an error the same way they would on a slightly older
kernel that doesn't support them yet. Most importantly, no unintended new ways
of performing dangerous operations will be silently allowed.

Another possible drawback is reduced system call performance due to the larger
filter created by the allowlist requiring more computation [2]. However, this
issue has not convincingly been demonstrated yet in practice, for example in
systemd or various browsers.

This commit tries to keep the behavior as close to unchanged as possible. Only
newer syscalls that are not supported by glibc 2.38 (as found in NixOS 23.11)
are blocked. Since this includes fchmodat2, the compatibility code added for
handling this syscall can be removed too.

[0] https://github.com/NixOS/nixpkgs/issues/300635
[1] https://github.com/NixOS/nix/issues/10424
[2] https://github.com/flatpak/flatpak/pull/4462#issuecomment-1061690607

Change-Id: I541be3ea9b249bcceddfed6a5a13ac10b11e16ad

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

@@ -87,6 +87,7 @@ 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

@@ -1,11 +0,0 @@
-#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

@@ -1,23 +0,0 @@
-#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,13 +393,10 @@ ref<eval_cache::EvalCache> openEvalCache(
     EvalState & state,
     std::shared_ptr<flake::LockedFlake> 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]()
+    auto fingerprint = evalSettings.useEvalCache && evalSettings.pureEval
+        ? std::make_optional(lockedFlake->getFingerprint())
+        : std::nullopt;
+    auto rootLoader = [&state, lockedFlake]()
         {
             /* For testing whether the evaluation cache is
                complete. */
@@ -415,7 +412,17 @@ 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,6 +1,5 @@
 libcmd_sources = files(
   'built-path.cc',
-  'command-installable-value.cc',
   'cmd-profiles.cc',
   'command.cc',
   'common-eval-args.cc',
@@ -18,7 +17,6 @@ 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,6 +33,10 @@ class EvalState;
 class StorePath;
 struct SingleDerivedPath;
 enum RepairFlag : bool;
+struct MemoryInputAccessor;
+namespace eval_cache {
+    class EvalCache;
+}
 
 
 /**
@@ -234,6 +238,11 @@ 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,547 +36,475 @@ 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);
 
-class ProgressBar : public Logger
+ProgressBar::~ProgressBar()
 {
-private:
+    stop();
+}
 
-    struct ActInfo
+/* Called by destructor, can't be overridden */
+void ProgressBar::stop()
+{
     {
-        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;
+        auto state(state_.lock());
+        if (!state->active) return;
+        state->active = false;
         writeToStderr("\r\e[K");
-    }
-
-    void resume() override {
-        state_.lock()->paused = false;
-        writeToStderr("\r\e[K");
-        state_.lock()->haveUpdate = true;
         updateCV.notify_one();
+        quitCV.notify_one();
+    }
+    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");
+        }
+        i->name = DrvName(name).name;
     }
 
-    bool isVerbose() override
-    {
-        return printBuildLogs;
+    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 log(Verbosity lvl, std::string_view s) override
-    {
-        if (lvl > verbosity) return;
-        auto state(state_.lock());
-        log(*state, lvl, s);
+    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;
     }
 
-    void logEI(const ErrorInfo & ei) override
-    {
-        auto state(state_.lock());
-
-        std::stringstream oss;
-        showErrorInfo(oss, ei, loggerSettings.showTrace.get());
-
-        log(*state, ei.level, oss.str());
+    if (type == actQueryPathInfo) {
+        auto name = storePathToName(getS(fields, 0));
+        i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1));
     }
 
-    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);
-        }
+    if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent))
+        || (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent))
+        || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)))
+        i->visible = false;
+
+    update(*state);
+}
+
+/* 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;
+}
+
+void ProgressBar::stopActivity(ActivityId act)
+{
+    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);
     }
 
-    void startActivity(ActivityId act, Verbosity lvl, ActivityType type,
-        const std::string & s, const Fields & fields, ActivityId parent) override
-    {
-        auto state(state_.lock());
+    update(*state);
+}
 
-        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");
-            }
-            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);
-        }
-
-        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 (type == actQueryPathInfo) {
-            auto name = storePathToName(getS(fields, 0));
-            i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1));
-        }
-
-        if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent))
-            || (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent))
-            || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)))
-            i->visible = false;
+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);
     }
 
-    /* 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 (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 {
-                    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->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());
-            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);
-        }
     }
 
-    void update(State & state)
-    {
-        state.haveUpdate = true;
-        updateCV.notify_one();
+    else if (type == resUntrustedPath) {
+        state->untrustedPaths++;
+        update(*state);
     }
 
-    std::chrono::milliseconds draw(State & state, const std::optional<std::string_view> & s)
-    {
-        auto nextWakeup = A_LONG_TIME;
+    else if (type == resCorruptedPath) {
+        state->corruptedPaths++;
+        update(*state);
+    }
 
-        state.haveUpdate = false;
-        if (state.paused || !state.active) return nextWakeup;
+    else if (type == resSetPhase) {
+        auto i = state->its.find(act);
+        assert(i != state->its.end());
+        i->second->phase = getS(fields, 0);
+        update(*state);
+    }
 
-        auto windowSize = getWindowSize();
-        auto width = windowSize.second;
-        if (width <= 0) {
-            width = std::numeric_limits<decltype(width)>::max();
-        }
+    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);
+    }
 
-        if (printMultiline && (state.lastLines >= 1)) {
-            // FIXME: make sure this works on windows
-            writeToStderr(fmt("\e[G\e[%dF\e[J", state.lastLines));
-        }
+    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);
+    }
+}
 
-        state.lastLines = 0;
+void ProgressBar::update(State & state)
+{
+    state.haveUpdate = true;
+    updateCV.notify_one();
+}
 
-        if (s != std::nullopt)
-            writeToStderr("\r\e[K" + filterANSIEscapes(s.value(), !isTTY) + ANSI_NORMAL "\n");
+std::chrono::milliseconds ProgressBar::draw(State & state, const std::optional<std::string_view> & s)
+{
+    auto nextWakeup = A_LONG_TIME;
 
-        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++;
-        }
+    state.haveUpdate = false;
+    if (state.paused || !state.active) return nextWakeup;
 
-        auto height = windowSize.first > 0 ? windowSize.first : 25;
-        auto moreBuilds = 0;
-        auto now = std::chrono::steady_clock::now();
+    auto windowSize = getWindowSize();
+    auto width = windowSize.second;
+    if (width <= 0) {
+        width = std::numeric_limits<decltype(width)>::max();
+    }
 
-        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;
-                }
+    if (printMultiline && (state.lastLines >= 1)) {
+        // FIXME: make sure this works on windows
+        writeToStderr(fmt("\e[G\e[%dF\e[J", state.lastLines));
+    }
 
-                /* 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)
+    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)
                         )
                     );
-                }
+            }
 
-                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 {
-                        moreBuilds++;
-                    }
-                }
+            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 (printMultiline && moreBuilds) {
-            writeToStderr(fmt("And %d more...", moreBuilds));
-        }
-
-        if (!printMultiline) {
-            line += " " + activity_line;
-            writeToStderr("\r" + filterANSIEscapes(line, false, width) + ANSI_NORMAL + "\e[K");
-        }
-
-        return nextWakeup;
     }
 
-    std::string getStatus(State & state)
-    {
-        auto MiB = 1024.0 * 1024.0;
+    if (printMultiline && moreActivities)
+        writeToStderr(fmt("And %d more...", moreActivities));
 
-        std::string res;
+    if (!printMultiline) {
+        line += " " + activity_line;
+         writeToStderr("\r" + filterANSIEscapes(line, false, width) + ANSI_NORMAL + "\e[K");
+    }
 
-        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;
-            }
+    return nextWakeup;
+}
 
-            expected = std::max(expected, act.expected);
+std::string ProgressBar::getStatus(State & state)
+{
+    constexpr auto MiB = 1024.0 * 1024.0;
 
-            std::string s;
+    std::string res;
 
-            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);
+    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;
+        }
+
+        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(done ? ANSI_GREEN + numberFmt + ANSI_NORMAL : numberFmt, done / unit);
-                s = fmt(itemFmt, s);
+                    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 (failed)
+                s += fmt(" (" ANSI_RED "%d failed" ANSI_NORMAL ")", failed / unit);
+        }
 
-            return s;
-        };
+        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;
+    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;
-        };
-
-        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);
         }
     }
 
-    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 {};
+    // 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);
         draw(*state, {});
-        return s[0];
+    } else {
+        Logger::writeToStdout(s);
     }
+}
 
-    void setPrintBuildLogs(bool printBuildLogs) override
-    {
-        this->printBuildLogs = printBuildLogs;
-    }
+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 setPrintMultiline(bool printMultiline) override
-    {
-        this->printMultiline = printMultiline;
-    }
-};
+void ProgressBar::setPrintBuildLogs(bool printBuildLogs)
+{
+    this->printBuildLogs = printBuildLogs;
+}
+
+void ProgressBar::setPrintMultiline(bool printMultiline)
+{
+    this->printMultiline = printMultiline;
+}
 
 Logger * makeProgressBar()
 {

src/libmain/progress-bar.hh

@@ -1,10 +1,135 @@
 #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 = getFile(cacheInfoFile);
+    auto cacheInfo = getFileContents(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(*getFile(path));
+    sink(*getFileContents(path));
 }
 
-std::optional<std::string> BinaryCacheStore::getFile(const std::string & path)
+std::optional<std::string> BinaryCacheStore::getFileContents(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 = getFile(narInfoFile);
+    auto data = getFileContents(narInfoFile);
 
     if (!data) return nullptr;
 
@@ -385,7 +385,7 @@ StorePath BinaryCacheStore::addToStore(
     if (method == FileIngestionMethod::Recursive) {
         dumpPath(srcPath, sink, filter);
     } else {
-        readFile(srcPath, sink);
+        readFileSource(srcPath)->drainInto(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 = getFile(outputInfoFilePath);
+    auto data = getFileContents(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 getFile(logPath);
+    return getFileContents(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> getFile(const std::string & path);
+    virtual std::optional<std::string> getFileContents(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 &) {
-                    readFile(actualPath, caSink);
+                    readFileSource(actualPath)->drainInto(caSink);
                 },
                 [&](const FileIngestionMethod & m2) {
                     switch (m2) {
@@ -2989,7 +2989,7 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
                         dumpPath(actualPath, caSink);
                         break;
                     case FileIngestionMethod::Flat:
-                        readFile(actualPath, caSink);
+                        readFileSource(actualPath)->drainInto(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), *decompressor);
+            fileTransfer->download(std::move(request))->drainInto(*decompressor);
             decompressor->finish();
         });
 

src/libstore/daemon.cc

@@ -296,17 +296,6 @@ 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();
@@ -316,36 +305,74 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
-    case WorkerProto::Op::QueryPathHash: {
-        auto path = store->parseStorePath(readString(from));
-        logger->startWork();
-        auto hash = store->queryPathInfo(path)->narHash;
-        logger->stopWork();
-        to << hash.to_string(Base16, false);
+    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)");
         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;
-        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);
+
+        #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
+
         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();
@@ -363,15 +390,6 @@ 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();
@@ -493,29 +511,6 @@ 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));
@@ -666,8 +661,10 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         break;
     }
 
-    // Obsolete.
+    // Obsolete since 9947f1646a26b339fff2e02b77798e9841fac7f0 (included in CppNix 2.5.0).
     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,16 +686,8 @@ struct curlFileTransfer : public FileTransfer
             ->callback.get_future();
     }
 
-    void download(FileTransferRequest && request, Sink & sink) override
+    box_ptr<Source> download(FileTransferRequest && request) 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;
@@ -705,13 +697,6 @@ 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) {
@@ -750,50 +735,99 @@ struct curlFileTransfer : public FileTransfer
             }
         );
 
-        std::unique_ptr<FinishSink> decompressor;
-
-        while (true) {
-            checkInterrupt();
-
+        struct InnerSource : Source
+        {
+            const std::shared_ptr<Sync<State>> _state;
             std::string chunk;
+            std::string_view buffered;
 
-            /* Grab data if available, otherwise wait for the download
-               thread to wake us up. */
+            explicit InnerSource(const std::shared_ptr<Sync<State>> & state) : _state(state) {}
+
+            ~InnerSource()
             {
+                // wake up the download thread if it's still going and have it abort
                 auto state(_state->lock());
-
-                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->failed |= !state->done;
                 state->request.notify_one();
             }
 
-            /* 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);
-        }
+            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;
     }
 };
 

src/libstore/filetransfer.hh

@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "box_ptr.hh"
 #include "logging.hh"
 #include "serialise.hh"
 #include "types.hh"
@@ -104,10 +105,13 @@ struct FileTransfer
     FileTransferResult transfer(const FileTransferRequest & request);
 
     /**
-     * Download a file, writing its data to a sink. The sink will be
-     * invoked on the thread of the caller.
+     * 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).
      */
-    virtual void download(FileTransferRequest && request, Sink & sink) = 0;
+    virtual box_ptr<Source> download(FileTransferRequest && request) = 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), sink);
+            getFileTransfer()->download(std::move(request))->drainInto(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> getFile(const std::string & path) override
+    std::optional<std::string> getFileContents(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 {
-            readFile(binaryCacheDir + "/" + path, sink);
+            readFileSource(binaryCacheDir + "/" + path)->drainInto(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 &) {
-            readFile(path, caSink);
+            readFileSource(path)->drainInto(caSink);
         },
         [&](const FileIngestionMethod & m2) {
             switch (m2) {
@@ -1898,7 +1898,7 @@ ContentAddress LocalStore::hashCAPath(
                 dumpPath(path, caSink);
                 break;
             case FileIngestionMethod::Flat:
-                readFile(path, caSink);
+                readFileSource(path)->drainInto(caSink);
                 break;
             }
         },

src/libstore/remote-store.cc

@@ -307,6 +307,7 @@ 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();
@@ -336,6 +337,7 @@ 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
@@ -530,6 +532,7 @@ 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 {
@@ -617,6 +620,7 @@ 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,6 +8,11 @@ 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
-            readFile(srcPath, sink);
+            readFileSource(srcPath)->drainInto(sink);
     });
     return addToStoreFromDump(*source, name, method, hashAlgo, repair, references);
 }
@@ -803,17 +803,31 @@ StorePathSet Store::queryValidPaths(const StorePathSet & paths, SubstituteFlag m
 
     auto doQuery = [&](const StorePath & path) {
         checkInterrupt();
-        auto state(state_.lock());
+
+        bool exists = false;
+        std::exception_ptr newExc{};
+
         try {
-            auto info = queryPathInfo(path);
-            state->valid.insert(path);
+            queryPathInfo(path);
+            exists = true;
         } catch (InvalidPath &) {
         } catch (...) {
-            state->exc = std::current_exception();
+            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();
         }
-        assert(state->left);
-        if (!--state->left)
-            wakeup.notify_one();
     };
 
     for (auto & path : paths)

src/libstore/worker-protocol.hh

@@ -23,6 +23,10 @@ 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
@@ -136,30 +140,30 @@ struct WorkerProto
 enum struct WorkerProto::Op : uint64_t
 {
     IsValidPath = 1,
-    HasSubstitutes = 3,
-    QueryPathHash = 4, // obsolete
-    QueryReferences = 5, // obsolete
+    HasSubstitutes = 3, // obsolete since 2012, stubbed to error
+    QueryPathHash = 4, // obsolete since 2016, stubbed to error
+    QueryReferences = 5, // obsolete since 2016, stubbed to error
     QueryReferrers = 6,
     AddToStore = 7,
-    AddTextToStore = 8, // obsolete since 1.25, Nix 3.0. Use WorkerProto::Op::AddToStore
+    AddTextToStore = 8, // obsolete since protocol 1.25, CppNix 2.4. Use WorkerProto::Op::AddToStore
     BuildPaths = 9,
     EnsurePath = 10,
     AddTempRoot = 11,
     AddIndirectRoot = 12,
-    SyncWithGC = 13,
+    SyncWithGC = 13, // obsolete since CppNix 2.5.0
     FindRoots = 14,
-    ExportPath = 16, // obsolete
-    QueryDeriver = 18, // obsolete
+    ExportPath = 16, // obsolete since 2017, stubbed to error
+    QueryDeriver = 18, // obsolete since 2016, stubbed to error
     SetOptions = 19,
     CollectGarbage = 20,
     QuerySubstitutablePathInfo = 21,
-    QueryDerivationOutputs = 22, // obsolete
+    QueryDerivationOutputs = 22, // obsolete since protocol 1.21, CppNix 2.4
     QueryAllValidPaths = 23,
-    QueryFailedPaths = 24,
-    ClearFailedPaths = 25,
+    QueryFailedPaths = 24, // obsolete, removed
+    ClearFailedPaths = 25, // obsolete, removed
     QueryPathInfo = 26,
-    ImportPaths = 27, // obsolete
-    QueryDerivationOutputNames = 28, // obsolete
+    ImportPaths = 27, // obsolete since 2016
+    QueryDerivationOutputNames = 28, // obsolete since CppNix 2.4
     QueryPathFromHashPart = 29,
     QuerySubstitutablePathInfos = 30,
     QueryValidPaths = 31,

src/libutil/compression.cc

@@ -163,23 +163,24 @@ struct BrotliDecompressionSource : Source
         uint8_t * out = (uint8_t *) data;
         const auto * begin = out;
 
-        try {
-            while (len && !BrotliDecoderIsFinished(state.get())) {
-                checkInterrupt();
+        while (len && !BrotliDecoderIsFinished(state.get())) {
+            checkInterrupt();
 
-                while (avail_in == 0) {
+            while (avail_in == 0) {
+                try {
                     avail_in = inner->read(buf.get(), BUF_SIZE);
-                    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");
+                } catch (EndOfFile &) {
+                    break;
                 }
+                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");
             }
-        } catch (EndOfFile &) {
         }
 
         if (begin != out) {
@@ -192,11 +193,9 @@ struct BrotliDecompressionSource : Source
 
 std::string decompress(const std::string & method, std::string_view in)
 {
-    StringSink ssink;
-    auto sink = makeDecompressionSink(method, ssink);
-    (*sink)(in);
-    sink->finish();
-    return std::move(ssink.s);
+    StringSource src{in};
+    auto filter = makeDecompressionSource(method, src);
+    return filter->drain();
 }
 
 std::unique_ptr<FinishSink> makeDecompressionSink(const std::string & method, Sink & nextSink)
@@ -224,6 +223,19 @@ 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,6 +19,7 @@ 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,12 +289,17 @@ std::string readFile(const Path & path)
 }
 
 
-void readFile(const Path & path, Sink & sink)
+box_ptr<Source> readFileSource(const Path & path)
 {
     AutoCloseFD fd{open(path.c_str(), O_RDONLY | O_CLOEXEC)};
     if (!fd)
         throw SysError("opening file '%s'", path);
-    drainFD(fd.get(), sink);
+
+    struct FileSource : FdSource {
+        AutoCloseFD fd;
+        explicit FileSource(AutoCloseFD fd) : FdSource(fd.get()), fd(std::move(fd)) {}
+    };
+    return make_box_ptr<FileSource>(std::move(fd));
 }
 
 

src/libutil/file-system.hh

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

src/nix/bundle.cc

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

src/nix/edit.cc

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

src/nix/eval.cc

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

src/nix/hash.cc

@@ -85,7 +85,7 @@ struct CmdHashBase : Command
 
             switch (mode) {
             case FileIngestionMethod::Flat:
-                readFile(path, *hashSink);
+                readFileSource(path)->drainInto(*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), sink);
+            getFileTransfer()->download(std::move(req))->drainInto(sink);
         }
 
         /* Optionally unpack the file. */

src/nix/run.cc

@@ -1,6 +1,7 @@
 #include "run.hh"
-#include "command-installable-value.hh"
+#include "command.hh"
 #include "common-args.hh"
+#include "installables.hh"
 #include "shared.hh"
 #include "store-api.hh"
 #include "derivations.hh"
@@ -145,7 +146,7 @@ struct CmdShell : InstallablesCommand, MixEnvironment
 
 static auto rCmdShell = registerCommand<CmdShell>("shell");
 
-struct CmdRun : InstallableValueCommand
+struct CmdRun : InstallableCommand
 {
     using InstallableCommand::run;
 
@@ -191,12 +192,14 @@ struct CmdRun : InstallableValueCommand
         return res;
     }
 
-    void run(ref<Store> store, ref<InstallableValue> installable) override
+    void run(ref<Store> store, ref<Installable> installable) override
     {
         auto state = getEvalState();
 
+        auto installableValue = InstallableValue::require(installable);
+
         lockFlags.applyNixConfig = true;
-        auto app = installable->toApp(*state).resolve(getEvalStore(), store);
+        auto app = installableValue->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-installable-value.hh"
+#include "command.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 : InstallableValueCommand, MixJSON
+struct CmdSearch : InstallableCommand, MixJSON
 {
     std::vector<std::string> res;
     std::vector<std::string> excludeRes;
@@ -62,8 +62,10 @@ struct CmdSearch : InstallableValueCommand, MixJSON
         };
     }
 
-    void run(ref<Store> store, ref<InstallableValue> installable) override
+    void run(ref<Store> store, ref<Installable> installable) override
     {
+        auto const installableValue = InstallableValue::require(installable);
+
         settings.readOnlyMode = true;
         evalSettings.enableImportFromDerivation.setDefault(false);
 
@@ -192,7 +194,7 @@ struct CmdSearch : InstallableValueCommand, MixJSON
             }
         };
 
-        for (auto & cursor : installable->getCursors(*state))
+        for (auto & cursor : installableValue->getCursors(*state))
             visit(*cursor, cursor->getAttrPath(), true);
 
         if (json)

tests/unit/libmain/progress-bar.cc

@@ -0,0 +1,43 @@
+#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,6 +7,7 @@
 #include <gtest/gtest.h>
 #include <netinet/in.h>
 #include <stdexcept>
+#include <string>
 #include <string_view>
 #include <sys/poll.h>
 #include <sys/socket.h>
@@ -136,7 +137,7 @@ TEST(FileTransfer, exceptionAbortsDownload)
 
     LambdaSink broken([](auto block) { throw Done(); });
 
-    ASSERT_THROW(ft->download(FileTransferRequest("file:///dev/zero"), broken), Done);
+    ASSERT_THROW(ft->download(FileTransferRequest("file:///dev/zero"))->drainInto(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
@@ -159,16 +160,21 @@ TEST(FileTransfer, NOT_ON_DARWIN(reportsSetupErrors))
         FileTransferError);
 }
 
-TEST(FileTransfer, NOT_ON_DARWIN(reportsTransferError))
+TEST(FileTransfer, NOT_ON_DARWIN(defersFailures))
 {
-    auto [port, srv] = serveHTTP("200 ok", "content-length: 100\r\n", [] {
+    auto [port, srv] = serveHTTP("200 ok", "content-length: 100000000\r\n", [] {
         std::this_thread::sleep_for(10ms);
-        return "";
+        // 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, ' ');
     });
     auto ft = makeFileTransfer();
     FileTransferRequest req(fmt("http://[::1]:%d/index", port));
     req.baseRetryTimeMs = 0;
-    ASSERT_THROW(ft->transfer(req), FileTransferError);
+    auto src = ft->download(std::move(req));
+    ASSERT_THROW(src->drain(), FileTransferError);
 }
 
 TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
@@ -180,7 +186,7 @@ TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
     auto ft = makeFileTransfer();
 
     StringSink sink;
-    ft->download(FileTransferRequest(fmt("http://[::1]:%d/index", port)), sink);
+    ft->download(FileTransferRequest(fmt("http://[::1]:%d/index", port)))->drainInto(sink);
     EXPECT_EQ(sink.s, original);
 }
 

tests/unit/libutil/compression.cc

@@ -66,6 +66,16 @@ 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
      * --------------------------------------------------------------------------*/
@@ -81,16 +91,17 @@ namespace nix {
     }
 
     TEST(makeCompressionSink, compressAndDecompress) {
-        StringSink strSink;
         auto inputString = "slfja;sljfklsa;jfklsjfkl;sdjfkl;sadjfkl;sdjf;lsdfjsadlf";
-        auto decompressionSink = makeDecompressionSink("bzip2", strSink);
-        auto sink = makeCompressionSink("bzip2", *decompressionSink);
 
+        StringSink strSink;
+        auto sink = makeCompressionSink("bzip2", strSink);
         (*sink)(inputString);
         sink->finish();
-        decompressionSink->finish();
 
-        ASSERT_STREQ(strSink.s.c_str(), inputString);
+        StringSource strSource{strSink.s};
+        auto decompressionSource = makeDecompressionSource("bzip2", strSource);
+
+        ASSERT_STREQ(decompressionSource->drain().c_str(), inputString);
     }
 
 }

tests/unit/meson.build

@@ -249,3 +249,25 @@ 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',
+)