Copy {uds-,}remote-store.{cc,hh}

This prepares for the splitting that happens in the next commit.

2 files changed, 1221 insertions, 0 deletions

diff --git a/src/libstore/uds-remote-store.cc b/src/libstore/uds-remote-store.cc
new file mode 100644
index 000000000..23b1942ce
--- /dev/null
+++ b/src/libstore/uds-remote-store.cc
@@ -0,0 +1,1017 @@
+#include "serialise.hh"
+#include "util.hh"
+#include "remote-fs-accessor.hh"
+#include "remote-store.hh"
+#include "worker-protocol.hh"
+#include "archive.hh"
+#include "affinity.hh"
+#include "globals.hh"
+#include "derivations.hh"
+#include "pool.hh"
+#include "finally.hh"
+#include "logging.hh"
+#include "callback.hh"
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <cstring>
+
+namespace nix {
+
+namespace worker_proto {
+
+std::string read(const Store & store, Source & from, Phantom<std::string> _)
+{
+    return readString(from);
+}
+
+void write(const Store & store, Sink & out, const std::string & str)
+{
+    out << str;
+}
+
+
+StorePath read(const Store & store, Source & from, Phantom<StorePath> _)
+{
+    return store.parseStorePath(readString(from));
+}
+
+void write(const Store & store, Sink & out, const StorePath & storePath)
+{
+    out << store.printStorePath(storePath);
+}
+
+
+ContentAddress read(const Store & store, Source & from, Phantom<ContentAddress> _)
+{
+    return parseContentAddress(readString(from));
+}
+
+void write(const Store & store, Sink & out, const ContentAddress & ca)
+{
+    out << renderContentAddress(ca);
+}
+
+
+std::optional<StorePath> read(const Store & store, Source & from, Phantom<std::optional<StorePath>> _)
+{
+    auto s = readString(from);
+    return s == "" ? std::optional<StorePath> {} : store.parseStorePath(s);
+}
+
+void write(const Store & store, Sink & out, const std::optional<StorePath> & storePathOpt)
+{
+    out << (storePathOpt ? store.printStorePath(*storePathOpt) : "");
+}
+
+
+std::optional<ContentAddress> read(const Store & store, Source & from, Phantom<std::optional<ContentAddress>> _)
+{
+    return parseContentAddressOpt(readString(from));
+}
+
+void write(const Store & store, Sink & out, const std::optional<ContentAddress> & caOpt)
+{
+    out << (caOpt ? renderContentAddress(*caOpt) : "");
+}
+
+}
+
+
+/* TODO: Separate these store impls into different files, give them better names */
+RemoteStore::RemoteStore(const Params & params)
+    : Store(params)
+    , RemoteStoreConfig(params)
+    , connections(make_ref<Pool<Connection>>(
+            std::max(1, (int) maxConnections),
+            [this]() {
+                auto conn = openConnectionWrapper();
+                try {
+                    initConnection(*conn);
+                } catch (...) {
+                    failed = true;
+                    throw;
+                }
+                return conn;
+            },
+            [this](const ref<Connection> & r) {
+                return
+                    r->to.good()
+                    && r->from.good()
+                    && std::chrono::duration_cast<std::chrono::seconds>(
+                        std::chrono::steady_clock::now() - r->startTime).count() < maxConnectionAge;
+            }
+            ))
+{
+}
+
+
+ref<RemoteStore::Connection> RemoteStore::openConnectionWrapper()
+{
+    if (failed)
+        throw Error("opening a connection to remote store '%s' previously failed", getUri());
+    try {
+        return openConnection();
+    } catch (...) {
+        failed = true;
+        throw;
+    }
+}
+
+
+UDSRemoteStore::UDSRemoteStore(const Params & params)
+    : StoreConfig(params)
+    , Store(params)
+    , LocalFSStore(params)
+    , RemoteStore(params)
+{
+}
+
+
+UDSRemoteStore::UDSRemoteStore(
+        const std::string scheme,
+        std::string socket_path,
+        const Params & params)
+    : UDSRemoteStore(params)
+{
+    path.emplace(socket_path);
+}
+
+
+std::string UDSRemoteStore::getUri()
+{
+    if (path) {
+        return std::string("unix://") + *path;
+    } else {
+        return "daemon";
+    }
+}
+
+
+ref<RemoteStore::Connection> UDSRemoteStore::openConnection()
+{
+    auto conn = make_ref<Connection>();
+
+    /* Connect to a daemon that does the privileged work for us. */
+    conn->fd = socket(PF_UNIX, SOCK_STREAM
+        #ifdef SOCK_CLOEXEC
+        | SOCK_CLOEXEC
+        #endif
+        , 0);
+    if (!conn->fd)
+        throw SysError("cannot create Unix domain socket");
+    closeOnExec(conn->fd.get());
+
+    string socketPath = path ? *path : settings.nixDaemonSocketFile;
+
+    struct sockaddr_un addr;
+    addr.sun_family = AF_UNIX;
+    if (socketPath.size() + 1 >= sizeof(addr.sun_path))
+        throw Error("socket path '%1%' is too long", socketPath);
+    strcpy(addr.sun_path, socketPath.c_str());
+
+    if (::connect(conn->fd.get(), (struct sockaddr *) &addr, sizeof(addr)) == -1)
+        throw SysError("cannot connect to daemon at '%1%'", socketPath);
+
+    conn->from.fd = conn->fd.get();
+    conn->to.fd = conn->fd.get();
+
+    conn->startTime = std::chrono::steady_clock::now();
+
+    return conn;
+}
+
+
+void RemoteStore::initConnection(Connection & conn)
+{
+    /* Send the magic greeting, check for the reply. */
+    try {
+        conn.to << WORKER_MAGIC_1;
+        conn.to.flush();
+        unsigned int magic = readInt(conn.from);
+        if (magic != WORKER_MAGIC_2) throw Error("protocol mismatch");
+
+        conn.from >> conn.daemonVersion;
+        if (GET_PROTOCOL_MAJOR(conn.daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION))
+            throw Error("Nix daemon protocol version not supported");
+        if (GET_PROTOCOL_MINOR(conn.daemonVersion) < 10)
+            throw Error("the Nix daemon version is too old");
+        conn.to << PROTOCOL_VERSION;
+
+        if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 14) {
+            int cpu = sameMachine() && settings.lockCPU ? lockToCurrentCPU() : -1;
+            if (cpu != -1)
+                conn.to << 1 << cpu;
+            else
+                conn.to << 0;
+        }
+
+        if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 11)
+            conn.to << false;
+
+        auto ex = conn.processStderr();
+        if (ex) std::rethrow_exception(ex);
+    }
+    catch (Error & e) {
+        throw Error("cannot open connection to remote store '%s': %s", getUri(), e.what());
+    }
+
+    setOptions(conn);
+}
+
+
+void RemoteStore::setOptions(Connection & conn)
+{
+    conn.to << wopSetOptions
+       << settings.keepFailed
+       << settings.keepGoing
+       << settings.tryFallback
+       << verbosity
+       << settings.maxBuildJobs
+       << settings.maxSilentTime
+       << true
+       << (settings.verboseBuild ? lvlError : lvlVomit)
+       << 0 // obsolete log type
+       << 0 /* obsolete print build trace */
+       << settings.buildCores
+       << settings.useSubstitutes;
+
+    if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 12) {
+        std::map<std::string, Config::SettingInfo> overrides;
+        globalConfig.getSettings(overrides, true);
+        overrides.erase(settings.keepFailed.name);
+        overrides.erase(settings.keepGoing.name);
+        overrides.erase(settings.tryFallback.name);
+        overrides.erase(settings.maxBuildJobs.name);
+        overrides.erase(settings.maxSilentTime.name);
+        overrides.erase(settings.buildCores.name);
+        overrides.erase(settings.useSubstitutes.name);
+        overrides.erase(loggerSettings.showTrace.name);
+        conn.to << overrides.size();
+        for (auto & i : overrides)
+            conn.to << i.first << i.second.value;
+    }
+
+    auto ex = conn.processStderr();
+    if (ex) std::rethrow_exception(ex);
+}
+
+
+/* A wrapper around Pool<RemoteStore::Connection>::Handle that marks
+   the connection as bad (causing it to be closed) if a non-daemon
+   exception is thrown before the handle is closed. Such an exception
+   causes a deviation from the expected protocol and therefore a
+   desynchronization between the client and daemon. */
+struct ConnectionHandle
+{
+    Pool<RemoteStore::Connection>::Handle handle;
+    bool daemonException = false;
+
+    ConnectionHandle(Pool<RemoteStore::Connection>::Handle && handle)
+        : handle(std::move(handle))
+    { }
+
+    ConnectionHandle(ConnectionHandle && h)
+        : handle(std::move(h.handle))
+    { }
+
+    ~ConnectionHandle()
+    {
+        if (!daemonException && std::uncaught_exceptions()) {
+            handle.markBad();
+            debug("closing daemon connection because of an exception");
+        }
+    }
+
+    RemoteStore::Connection * operator -> () { return &*handle; }
+
+    void processStderr(Sink * sink = 0, Source * source = 0, bool flush = true)
+    {
+        auto ex = handle->processStderr(sink, source, flush);
+        if (ex) {
+            daemonException = true;
+            std::rethrow_exception(ex);
+        }
+    }
+
+    void withFramedSink(std::function<void(Sink & sink)> fun);
+};
+
+
+ConnectionHandle RemoteStore::getConnection()
+{
+    return ConnectionHandle(connections->get());
+}
+
+
+bool RemoteStore::isValidPathUncached(const StorePath & path)
+{
+    auto conn(getConnection());
+    conn->to << wopIsValidPath << printStorePath(path);
+    conn.processStderr();
+    return readInt(conn->from);
+}
+
+
+StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
+{
+    auto conn(getConnection());
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
+        StorePathSet res;
+        for (auto & i : paths)
+            if (isValidPath(i)) res.insert(i);
+        return res;
+    } else {
+        conn->to << wopQueryValidPaths;
+        worker_proto::write(*this, conn->to, paths);
+        conn.processStderr();
+        return worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+    }
+}
+
+
+StorePathSet RemoteStore::queryAllValidPaths()
+{
+    auto conn(getConnection());
+    conn->to << wopQueryAllValidPaths;
+    conn.processStderr();
+    return worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+}
+
+
+StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths)
+{
+    auto conn(getConnection());
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
+        StorePathSet res;
+        for (auto & i : paths) {
+            conn->to << wopHasSubstitutes << printStorePath(i);
+            conn.processStderr();
+            if (readInt(conn->from)) res.insert(i);
+        }
+        return res;
+    } else {
+        conn->to << wopQuerySubstitutablePaths;
+        worker_proto::write(*this, conn->to, paths);
+        conn.processStderr();
+        return worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+    }
+}
+
+
+void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, SubstitutablePathInfos & infos)
+{
+    if (pathsMap.empty()) return;
+
+    auto conn(getConnection());
+
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 12) {
+
+        for (auto & i : pathsMap) {
+            SubstitutablePathInfo info;
+            conn->to << wopQuerySubstitutablePathInfo << printStorePath(i.first);
+            conn.processStderr();
+            unsigned int reply = readInt(conn->from);
+            if (reply == 0) continue;
+            auto deriver = readString(conn->from);
+            if (deriver != "")
+                info.deriver = parseStorePath(deriver);
+            info.references = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+            info.downloadSize = readLongLong(conn->from);
+            info.narSize = readLongLong(conn->from);
+            infos.insert_or_assign(i.first, std::move(info));
+        }
+
+    } else {
+
+        conn->to << wopQuerySubstitutablePathInfos;
+        if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 22) {
+            StorePathSet paths;
+            for (auto & path : pathsMap)
+                paths.insert(path.first);
+            worker_proto::write(*this, conn->to, paths);
+        } else
+            worker_proto::write(*this, conn->to, pathsMap);
+        conn.processStderr();
+        size_t count = readNum<size_t>(conn->from);
+        for (size_t n = 0; n < count; n++) {
+            SubstitutablePathInfo & info(infos[parseStorePath(readString(conn->from))]);
+            auto deriver = readString(conn->from);
+            if (deriver != "")
+                info.deriver = parseStorePath(deriver);
+            info.references = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+            info.downloadSize = readLongLong(conn->from);
+            info.narSize = readLongLong(conn->from);
+        }
+
+    }
+}
+
+
+ref<const ValidPathInfo> RemoteStore::readValidPathInfo(ConnectionHandle & conn, const StorePath & path)
+{
+    auto deriver = readString(conn->from);
+    auto narHash = Hash::parseAny(readString(conn->from), htSHA256);
+    auto info = make_ref<ValidPathInfo>(path, narHash);
+    if (deriver != "") info->deriver = parseStorePath(deriver);
+    info->references = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+    conn->from >> info->registrationTime >> info->narSize;
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 16) {
+        conn->from >> info->ultimate;
+        info->sigs = readStrings<StringSet>(conn->from);
+        info->ca = parseContentAddressOpt(readString(conn->from));
+    }
+    return info;
+}
+
+
+void RemoteStore::queryPathInfoUncached(const StorePath & path,
+    Callback<std::shared_ptr<const ValidPathInfo>> callback) noexcept
+{
+    try {
+        std::shared_ptr<const ValidPathInfo> info;
+        {
+            auto conn(getConnection());
+            conn->to << wopQueryPathInfo << printStorePath(path);
+            try {
+                conn.processStderr();
+            } catch (Error & e) {
+                // Ugly backwards compatibility hack.
+                if (e.msg().find("is not valid") != std::string::npos)
+                    throw InvalidPath(e.info());
+                throw;
+            }
+            if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 17) {
+                bool valid; conn->from >> valid;
+                if (!valid) throw InvalidPath("path '%s' is not valid", printStorePath(path));
+            }
+            info = readValidPathInfo(conn, path);
+        }
+        callback(std::move(info));
+    } catch (...) { callback.rethrow(); }
+}
+
+
+void RemoteStore::queryReferrers(const StorePath & path,
+    StorePathSet & referrers)
+{
+    auto conn(getConnection());
+    conn->to << wopQueryReferrers << printStorePath(path);
+    conn.processStderr();
+    for (auto & i : worker_proto::read(*this, conn->from, Phantom<StorePathSet> {}))
+        referrers.insert(i);
+}
+
+
+StorePathSet RemoteStore::queryValidDerivers(const StorePath & path)
+{
+    auto conn(getConnection());
+    conn->to << wopQueryValidDerivers << printStorePath(path);
+    conn.processStderr();
+    return worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+}
+
+
+StorePathSet RemoteStore::queryDerivationOutputs(const StorePath & path)
+{
+    auto conn(getConnection());
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 0x16) {
+        return Store::queryDerivationOutputs(path);
+    }
+    conn->to << wopQueryDerivationOutputs << printStorePath(path);
+    conn.processStderr();
+    return worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+}
+
+
+std::map<std::string, std::optional<StorePath>> RemoteStore::queryPartialDerivationOutputMap(const StorePath & path)
+{
+    if (GET_PROTOCOL_MINOR(getProtocol()) >= 0x16) {
+        auto conn(getConnection());
+        conn->to << wopQueryDerivationOutputMap << printStorePath(path);
+        conn.processStderr();
+        return worker_proto::read(*this, conn->from, Phantom<std::map<std::string, std::optional<StorePath>>> {});
+    } else {
+        // Fallback for old daemon versions.
+        // For floating-CA derivations (and their co-dependencies) this is an
+        // under-approximation as it only returns the paths that can be inferred
+        // from the derivation itself (and not the ones that are known because
+        // the have been built), but as old stores don't handle floating-CA
+        // derivations this shouldn't matter
+        auto derivation = readDerivation(path);
+        auto outputsWithOptPaths = derivation.outputsAndOptPaths(*this);
+        std::map<std::string, std::optional<StorePath>> ret;
+        for (auto & [outputName, outputAndPath] : outputsWithOptPaths) {
+            ret.emplace(outputName, outputAndPath.second);
+        }
+        return ret;
+    }
+}
+
+std::optional<StorePath> RemoteStore::queryPathFromHashPart(const std::string & hashPart)
+{
+    auto conn(getConnection());
+    conn->to << wopQueryPathFromHashPart << hashPart;
+    conn.processStderr();
+    Path path = readString(conn->from);
+    if (path.empty()) return {};
+    return parseStorePath(path);
+}
+
+
+ref<const ValidPathInfo> RemoteStore::addCAToStore(
+    Source & dump,
+    const string & name,
+    ContentAddressMethod caMethod,
+    const StorePathSet & references,
+    RepairFlag repair)
+{
+    std::optional<ConnectionHandle> conn_(getConnection());
+    auto & conn = *conn_;
+
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) {
+
+        conn->to
+            << wopAddToStore
+            << name
+            << renderContentAddressMethod(caMethod);
+        worker_proto::write(*this, conn->to, references);
+        conn->to << repair;
+
+        conn.withFramedSink([&](Sink & sink) {
+            dump.drainInto(sink);
+        });
+
+        auto path = parseStorePath(readString(conn->from));
+        return readValidPathInfo(conn, path);
+    }
+    else {
+        if (repair) throw Error("repairing is not supported when building through the Nix daemon protocol < 1.25");
+
+        std::visit(overloaded {
+            [&](TextHashMethod thm) -> void {
+                std::string s = dump.drain();
+                conn->to << wopAddTextToStore << name << s;
+                worker_proto::write(*this, conn->to, references);
+                conn.processStderr();
+            },
+            [&](FixedOutputHashMethod fohm) -> void {
+                conn->to
+                    << wopAddToStore
+                    << name
+                    << ((fohm.hashType == htSHA256 && fohm.fileIngestionMethod == FileIngestionMethod::Recursive) ? 0 : 1) /* backwards compatibility hack */
+                    << (fohm.fileIngestionMethod == FileIngestionMethod::Recursive ? 1 : 0)
+                    << printHashType(fohm.hashType);
+
+                try {
+                    conn->to.written = 0;
+                    conn->to.warn = true;
+                    connections->incCapacity();
+                    {
+                        Finally cleanup([&]() { connections->decCapacity(); });
+                        if (fohm.fileIngestionMethod == FileIngestionMethod::Recursive) {
+                            dump.drainInto(conn->to);
+                        } else {
+                            std::string contents = dump.drain();
+                            dumpString(contents, conn->to);
+                        }
+                    }
+                    conn->to.warn = false;
+                    conn.processStderr();
+                } catch (SysError & e) {
+                    /* Daemon closed while we were sending the path. Probably OOM
+                      or I/O error. */
+                    if (e.errNo == EPIPE)
+                        try {
+                            conn.processStderr();
+                        } catch (EndOfFile & e) { }
+                    throw;
+                }
+
+            }
+        }, caMethod);
+        auto path = parseStorePath(readString(conn->from));
+        // Release our connection to prevent a deadlock in queryPathInfo().
+        conn_.reset();
+        return queryPathInfo(path);
+    }
+}
+
+
+StorePath RemoteStore::addToStoreFromDump(Source & dump, const string & name,
+        FileIngestionMethod method, HashType hashType, RepairFlag repair)
+{
+    StorePathSet references;
+    return addCAToStore(dump, name, FixedOutputHashMethod{ .fileIngestionMethod = method, .hashType = hashType }, references, repair)->path;
+}
+
+
+void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
+    RepairFlag repair, CheckSigsFlag checkSigs)
+{
+    auto conn(getConnection());
+
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 18) {
+        conn->to << wopImportPaths;
+
+        auto source2 = sinkToSource([&](Sink & sink) {
+            sink << 1 // == path follows
+                ;
+            copyNAR(source, sink);
+            sink
+                << exportMagic
+                << printStorePath(info.path);
+            worker_proto::write(*this, sink, info.references);
+            sink
+                << (info.deriver ? printStorePath(*info.deriver) : "")
+                << 0 // == no legacy signature
+                << 0 // == no path follows
+                ;
+        });
+
+        conn.processStderr(0, source2.get());
+
+        auto importedPaths = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+        assert(importedPaths.size() <= 1);
+    }
+
+    else {
+        conn->to << wopAddToStoreNar
+                 << printStorePath(info.path)
+                 << (info.deriver ? printStorePath(*info.deriver) : "")
+                 << info.narHash.to_string(Base16, false);
+        worker_proto::write(*this, conn->to, info.references);
+        conn->to << info.registrationTime << info.narSize
+                 << info.ultimate << info.sigs << renderContentAddress(info.ca)
+                 << repair << !checkSigs;
+
+        if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 23) {
+            conn.withFramedSink([&](Sink & sink) {
+                copyNAR(source, sink);
+            });
+        } else if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 21) {
+            conn.processStderr(0, &source);
+        } else {
+            copyNAR(source, conn->to);
+            conn.processStderr(0, nullptr);
+        }
+    }
+}
+
+
+StorePath RemoteStore::addTextToStore(const string & name, const string & s,
+    const StorePathSet & references, RepairFlag repair)
+{
+    StringSource source(s);
+    return addCAToStore(source, name, TextHashMethod{}, references, repair)->path;
+}
+
+
+void RemoteStore::buildPaths(const std::vector<StorePathWithOutputs> & drvPaths, BuildMode buildMode)
+{
+    auto conn(getConnection());
+    conn->to << wopBuildPaths;
+    assert(GET_PROTOCOL_MINOR(conn->daemonVersion) >= 13);
+    Strings ss;
+    for (auto & p : drvPaths)
+        ss.push_back(p.to_string(*this));
+    conn->to << ss;
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 15)
+        conn->to << buildMode;
+    else
+        /* Old daemons did not take a 'buildMode' parameter, so we
+           need to validate it here on the client side.  */
+        if (buildMode != bmNormal)
+            throw Error("repairing or checking is not supported when building through the Nix daemon");
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+BuildResult RemoteStore::buildDerivation(const StorePath & drvPath, const BasicDerivation & drv,
+    BuildMode buildMode)
+{
+    auto conn(getConnection());
+    conn->to << wopBuildDerivation << printStorePath(drvPath);
+    writeDerivation(conn->to, *this, drv);
+    conn->to << buildMode;
+    conn.processStderr();
+    BuildResult res;
+    unsigned int status;
+    conn->from >> status >> res.errorMsg;
+    res.status = (BuildResult::Status) status;
+    return res;
+}
+
+
+void RemoteStore::ensurePath(const StorePath & path)
+{
+    auto conn(getConnection());
+    conn->to << wopEnsurePath << printStorePath(path);
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+void RemoteStore::addTempRoot(const StorePath & path)
+{
+    auto conn(getConnection());
+    conn->to << wopAddTempRoot << printStorePath(path);
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+void RemoteStore::addIndirectRoot(const Path & path)
+{
+    auto conn(getConnection());
+    conn->to << wopAddIndirectRoot << path;
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+void RemoteStore::syncWithGC()
+{
+    auto conn(getConnection());
+    conn->to << wopSyncWithGC;
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+Roots RemoteStore::findRoots(bool censor)
+{
+    auto conn(getConnection());
+    conn->to << wopFindRoots;
+    conn.processStderr();
+    size_t count = readNum<size_t>(conn->from);
+    Roots result;
+    while (count--) {
+        Path link = readString(conn->from);
+        auto target = parseStorePath(readString(conn->from));
+        result[std::move(target)].emplace(link);
+    }
+    return result;
+}
+
+
+void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
+{
+    auto conn(getConnection());
+
+    conn->to
+        << wopCollectGarbage << options.action;
+    worker_proto::write(*this, conn->to, options.pathsToDelete);
+    conn->to << options.ignoreLiveness
+        << options.maxFreed
+        /* removed options */
+        << 0 << 0 << 0;
+
+    conn.processStderr();
+
+    results.paths = readStrings<PathSet>(conn->from);
+    results.bytesFreed = readLongLong(conn->from);
+    readLongLong(conn->from); // obsolete
+
+    {
+        auto state_(Store::state.lock());
+        state_->pathInfoCache.clear();
+    }
+}
+
+
+void RemoteStore::optimiseStore()
+{
+    auto conn(getConnection());
+    conn->to << wopOptimiseStore;
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+bool RemoteStore::verifyStore(bool checkContents, RepairFlag repair)
+{
+    auto conn(getConnection());
+    conn->to << wopVerifyStore << checkContents << repair;
+    conn.processStderr();
+    return readInt(conn->from);
+}
+
+
+void RemoteStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
+{
+    auto conn(getConnection());
+    conn->to << wopAddSignatures << printStorePath(storePath) << sigs;
+    conn.processStderr();
+    readInt(conn->from);
+}
+
+
+void RemoteStore::queryMissing(const std::vector<StorePathWithOutputs> & targets,
+    StorePathSet & willBuild, StorePathSet & willSubstitute, StorePathSet & unknown,
+    uint64_t & downloadSize, uint64_t & narSize)
+{
+    {
+        auto conn(getConnection());
+        if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 19)
+            // Don't hold the connection handle in the fallback case
+            // to prevent a deadlock.
+            goto fallback;
+        conn->to << wopQueryMissing;
+        Strings ss;
+        for (auto & p : targets)
+            ss.push_back(p.to_string(*this));
+        conn->to << ss;
+        conn.processStderr();
+        willBuild = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+        willSubstitute = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+        unknown = worker_proto::read(*this, conn->from, Phantom<StorePathSet> {});
+        conn->from >> downloadSize >> narSize;
+        return;
+    }
+
+ fallback:
+    return Store::queryMissing(targets, willBuild, willSubstitute,
+        unknown, downloadSize, narSize);
+}
+
+
+void RemoteStore::connect()
+{
+    auto conn(getConnection());
+}
+
+
+unsigned int RemoteStore::getProtocol()
+{
+    auto conn(connections->get());
+    return conn->daemonVersion;
+}
+
+
+void RemoteStore::flushBadConnections()
+{
+    connections->flushBad();
+}
+
+
+RemoteStore::Connection::~Connection()
+{
+    try {
+        to.flush();
+    } catch (...) {
+        ignoreException();
+    }
+}
+
+void RemoteStore::narFromPath(const StorePath & path, Sink & sink)
+{
+    auto conn(connections->get());
+    conn->to << wopNarFromPath << printStorePath(path);
+    conn->processStderr();
+    copyNAR(conn->from, sink);
+}
+
+ref<FSAccessor> RemoteStore::getFSAccessor()
+{
+    return make_ref<RemoteFSAccessor>(ref<Store>(shared_from_this()));
+}
+
+static Logger::Fields readFields(Source & from)
+{
+    Logger::Fields fields;
+    size_t size = readInt(from);
+    for (size_t n = 0; n < size; n++) {
+        auto type = (decltype(Logger::Field::type)) readInt(from);
+        if (type == Logger::Field::tInt)
+            fields.push_back(readNum<uint64_t>(from));
+        else if (type == Logger::Field::tString)
+            fields.push_back(readString(from));
+        else
+            throw Error("got unsupported field type %x from Nix daemon", (int) type);
+    }
+    return fields;
+}
+
+
+std::exception_ptr RemoteStore::Connection::processStderr(Sink * sink, Source * source, bool flush)
+{
+    if (flush)
+        to.flush();
+
+    while (true) {
+
+        auto msg = readNum<uint64_t>(from);
+
+        if (msg == STDERR_WRITE) {
+            string s = readString(from);
+            if (!sink) throw Error("no sink");
+            (*sink)(s);
+        }
+
+        else if (msg == STDERR_READ) {
+            if (!source) throw Error("no source");
+            size_t len = readNum<size_t>(from);
+            auto buf = std::make_unique<unsigned char[]>(len);
+            writeString(buf.get(), source->read(buf.get(), len), to);
+            to.flush();
+        }
+
+        else if (msg == STDERR_ERROR) {
+            if (GET_PROTOCOL_MINOR(daemonVersion) >= 26) {
+                return std::make_exception_ptr(readError(from));
+            } else {
+                string error = readString(from);
+                unsigned int status = readInt(from);
+                return std::make_exception_ptr(Error(status, error));
+            }
+        }
+
+        else if (msg == STDERR_NEXT)
+            printError(chomp(readString(from)));
+
+        else if (msg == STDERR_START_ACTIVITY) {
+            auto act = readNum<ActivityId>(from);
+            auto lvl = (Verbosity) readInt(from);
+            auto type = (ActivityType) readInt(from);
+            auto s = readString(from);
+            auto fields = readFields(from);
+            auto parent = readNum<ActivityId>(from);
+            logger->startActivity(act, lvl, type, s, fields, parent);
+        }
+
+        else if (msg == STDERR_STOP_ACTIVITY) {
+            auto act = readNum<ActivityId>(from);
+            logger->stopActivity(act);
+        }
+
+        else if (msg == STDERR_RESULT) {
+            auto act = readNum<ActivityId>(from);
+            auto type = (ResultType) readInt(from);
+            auto fields = readFields(from);
+            logger->result(act, type, fields);
+        }
+
+        else if (msg == STDERR_LAST)
+            break;
+
+        else
+            throw Error("got unknown message type %x from Nix daemon", msg);
+    }
+
+    return nullptr;
+}
+
+void ConnectionHandle::withFramedSink(std::function<void(Sink &sink)> fun)
+{
+    (*this)->to.flush();
+
+    std::exception_ptr ex;
+
+    /* Handle log messages / exceptions from the remote on a
+        separate thread. */
+    std::thread stderrThread([&]()
+    {
+        try {
+            processStderr(nullptr, nullptr, false);
+        } catch (...) {
+            ex = std::current_exception();
+        }
+    });
+
+    Finally joinStderrThread([&]()
+    {
+        if (stderrThread.joinable()) {
+            stderrThread.join();
+            if (ex) {
+                try {
+                    std::rethrow_exception(ex);
+                } catch (...) {
+                    ignoreException();
+                }
+            }
+        }
+    });
+
+    {
+        FramedSink sink((*this)->to, ex);
+        fun(sink);
+        sink.flush();
+    }
+
+    stderrThread.join();
+    if (ex)
+        std::rethrow_exception(ex);
+
+}
+
+static RegisterStoreImplementation<UDSRemoteStore, UDSRemoteStoreConfig> regUDSRemoteStore;
+
+}
diff --git a/src/libstore/uds-remote-store.hh b/src/libstore/uds-remote-store.hh
new file mode 100644
index 000000000..554fb6bed
--- /dev/null
+++ b/src/libstore/uds-remote-store.hh
@@ -0,0 +1,204 @@
+#pragma once
+
+#include <limits>
+#include <string>
+
+#include "store-api.hh"
+#include "local-fs-store.hh"
+
+
+namespace nix {
+
+
+class Pipe;
+class Pid;
+struct FdSink;
+struct FdSource;
+template<typename T> class Pool;
+struct ConnectionHandle;
+
+struct RemoteStoreConfig : virtual StoreConfig
+{
+    using StoreConfig::StoreConfig;
+
+    const Setting<int> maxConnections{(StoreConfig*) this, 1,
+            "max-connections", "maximum number of concurrent connections to the Nix daemon"};
+
+    const Setting<unsigned int> maxConnectionAge{(StoreConfig*) this, std::numeric_limits<unsigned int>::max(),
+            "max-connection-age", "number of seconds to reuse a connection"};
+};
+
+/* FIXME: RemoteStore is a misnomer - should be something like
+   DaemonStore. */
+class RemoteStore : public virtual Store, public virtual RemoteStoreConfig
+{
+public:
+
+    virtual bool sameMachine() = 0;
+
+    RemoteStore(const Params & params);
+
+    /* Implementations of abstract store API methods. */
+
+    bool isValidPathUncached(const StorePath & path) override;
+
+    StorePathSet queryValidPaths(const StorePathSet & paths,
+        SubstituteFlag maybeSubstitute = NoSubstitute) override;
+
+    StorePathSet queryAllValidPaths() override;
+
+    void queryPathInfoUncached(const StorePath & path,
+        Callback<std::shared_ptr<const ValidPathInfo>> callback) noexcept override;
+
+    void queryReferrers(const StorePath & path, StorePathSet & referrers) override;
+
+    StorePathSet queryValidDerivers(const StorePath & path) override;
+
+    StorePathSet queryDerivationOutputs(const StorePath & path) override;
+
+    std::map<std::string, std::optional<StorePath>> queryPartialDerivationOutputMap(const StorePath & path) override;
+    std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override;
+
+    StorePathSet querySubstitutablePaths(const StorePathSet & paths) override;
+
+    void querySubstitutablePathInfos(const StorePathCAMap & paths,
+        SubstitutablePathInfos & infos) override;
+
+    /* Add a content-addressable store path. `dump` will be drained. */
+    ref<const ValidPathInfo> addCAToStore(
+        Source & dump,
+        const string & name,
+        ContentAddressMethod caMethod,
+        const StorePathSet & references,
+        RepairFlag repair);
+
+    /* Add a content-addressable store path. Does not support references. `dump` will be drained. */
+    StorePath addToStoreFromDump(Source & dump, const string & name,
+        FileIngestionMethod method = FileIngestionMethod::Recursive, HashType hashAlgo = htSHA256, RepairFlag repair = NoRepair) override;
+
+    void addToStore(const ValidPathInfo & info, Source & nar,
+        RepairFlag repair, CheckSigsFlag checkSigs) override;
+
+    StorePath addTextToStore(const string & name, const string & s,
+        const StorePathSet & references, RepairFlag repair) override;
+
+    void buildPaths(const std::vector<StorePathWithOutputs> & paths, BuildMode buildMode) override;
+
+    BuildResult buildDerivation(const StorePath & drvPath, const BasicDerivation & drv,
+        BuildMode buildMode) override;
+
+    void ensurePath(const StorePath & path) override;
+
+    void addTempRoot(const StorePath & path) override;
+
+    void addIndirectRoot(const Path & path) override;
+
+    void syncWithGC() override;
+
+    Roots findRoots(bool censor) override;
+
+    void collectGarbage(const GCOptions & options, GCResults & results) override;
+
+    void optimiseStore() override;
+
+    bool verifyStore(bool checkContents, RepairFlag repair) override;
+
+    void addSignatures(const StorePath & storePath, const StringSet & sigs) override;
+
+    void queryMissing(const std::vector<StorePathWithOutputs> & targets,
+        StorePathSet & willBuild, StorePathSet & willSubstitute, StorePathSet & unknown,
+        uint64_t & downloadSize, uint64_t & narSize) override;
+
+    void connect() override;
+
+    unsigned int getProtocol() override;
+
+    void flushBadConnections();
+
+    struct Connection
+    {
+        AutoCloseFD fd;
+        FdSink to;
+        FdSource from;
+        unsigned int daemonVersion;
+        std::chrono::time_point<std::chrono::steady_clock> startTime;
+
+        virtual ~Connection();
+
+        std::exception_ptr processStderr(Sink * sink = 0, Source * source = 0, bool flush = true);
+    };
+
+    ref<Connection> openConnectionWrapper();
+
+protected:
+
+    virtual ref<Connection> openConnection() = 0;
+
+    void initConnection(Connection & conn);
+
+    ref<Pool<Connection>> connections;
+
+    virtual void setOptions(Connection & conn);
+
+    ConnectionHandle getConnection();
+
+    friend struct ConnectionHandle;
+
+    virtual ref<FSAccessor> getFSAccessor() override;
+
+    virtual void narFromPath(const StorePath & path, Sink & sink) override;
+
+    ref<const ValidPathInfo> readValidPathInfo(ConnectionHandle & conn, const StorePath & path);
+
+private:
+
+    std::atomic_bool failed{false};
+
+};
+
+struct UDSRemoteStoreConfig : virtual LocalFSStoreConfig, virtual RemoteStoreConfig
+{
+    UDSRemoteStoreConfig(const Store::Params & params)
+        : StoreConfig(params)
+        , LocalFSStoreConfig(params)
+        , RemoteStoreConfig(params)
+    {
+    }
+
+    UDSRemoteStoreConfig()
+        : UDSRemoteStoreConfig(Store::Params({}))
+    {
+    }
+
+    const std::string name() override { return "Local Daemon Store"; }
+};
+
+class UDSRemoteStore : public LocalFSStore, public RemoteStore, public virtual UDSRemoteStoreConfig
+{
+public:
+
+    UDSRemoteStore(const Params & params);
+    UDSRemoteStore(const std::string scheme, std::string path, const Params & params);
+
+    std::string getUri() override;
+
+    static std::set<std::string> uriSchemes()
+    { return {"unix"}; }
+
+    bool sameMachine() override
+    { return true; }
+
+    ref<FSAccessor> getFSAccessor() override
+    { return LocalFSStore::getFSAccessor(); }
+
+    void narFromPath(const StorePath & path, Sink & sink) override
+    { LocalFSStore::narFromPath(path, sink); }
+
+private:
+
+    ref<RemoteStore::Connection> openConnection() override;
+    std::optional<std::string> path;
+};
+
+
+}