diff --git a/flake.lock b/flake.lock
index 2ddc7c16..d8b2e4a9 100644
--- a/flake.lock
+++ b/flake.lock
@@ -16,58 +16,58 @@
         "type": "github"
       }
     },
-    "lowdown-src": {
+    "libgit2": {
       "flake": false,
       "locked": {
-        "lastModified": 1633514407,
-        "narHash": "sha256-Dw32tiMjdK9t3ETl5fzGrutQTzh2rufgZV4A/BbxuD4=",
-        "owner": "kristapsdz",
-        "repo": "lowdown",
-        "rev": "d2c2b44ff6c27b936ec27358a2653caaef8f73b8",
+        "lastModified": 1697646580,
+        "narHash": "sha256-oX4Z3S9WtJlwvj0uH9HlYcWv+x1hqp8mhXl7HsLu2f0=",
+        "owner": "libgit2",
+        "repo": "libgit2",
+        "rev": "45fd9ed7ae1a9b74b957ef4f337bc3c8b3df01b5",
         "type": "github"
       },
       "original": {
-        "owner": "kristapsdz",
-        "repo": "lowdown",
+        "owner": "libgit2",
+        "repo": "libgit2",
         "type": "github"
       }
     },
     "nix": {
       "inputs": {
         "flake-compat": "flake-compat",
-        "lowdown-src": "lowdown-src",
+        "libgit2": "libgit2",
         "nixpkgs": [
           "nixpkgs"
         ],
         "nixpkgs-regression": "nixpkgs-regression"
       },
       "locked": {
-        "lastModified": 1706208340,
-        "narHash": "sha256-wNyHUEIiKKVs6UXrUzhP7RSJQv0A8jckgcuylzftl8k=",
+        "lastModified": 1706637536,
+        "narHash": "sha256-fjx+nCOzuSxGWfhwWWc8hCsLFZAjZLDDUcbBtldRqbk=",
         "owner": "NixOS",
         "repo": "nix",
-        "rev": "2c4bb93ba5a97e7078896ebc36385ce172960e4e",
+        "rev": "8f42912c80c0a03f62f6a3d28a3af05a9762565d",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "2.19-maintenance",
+        "ref": "2.20-maintenance",
         "repo": "nix",
         "type": "github"
       }
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1701615100,
-        "narHash": "sha256-7VI84NGBvlCTduw2aHLVB62NvCiZUlALLqBe5v684Aw=",
+        "lastModified": 1705033721,
+        "narHash": "sha256-K5eJHmL1/kev6WuqyqqbS1cdNnSidIZ3jeqJ7GbrYnQ=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "e9f06adb793d1cca5384907b3b8a4071d5d7cb19",
+        "rev": "a1982c92d8980a0114372973cbdfe0a307f1bdea",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-23.05",
+        "ref": "nixos-23.05-small",
         "repo": "nixpkgs",
         "type": "github"
       }
diff --git a/flake.nix b/flake.nix
index 30aa61c0..306ed292 100644
--- a/flake.nix
+++ b/flake.nix
@@ -1,8 +1,8 @@
 {
   description = "A Nix-based continuous build system";
 
-  inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-23.05";
-  inputs.nix.url = "github:NixOS/nix/2.19-maintenance";
+  inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-23.05-small";
+  inputs.nix.url = "github:NixOS/nix/2.20-maintenance";
   inputs.nix.inputs.nixpkgs.follows = "nixpkgs";
 
   # TODO get rid of this once https://github.com/NixOS/nix/pull/9546 is
diff --git a/src/hydra-eval-jobs/hydra-eval-jobs.cc b/src/hydra-eval-jobs/hydra-eval-jobs.cc
index 5cfc3a52..1d7de74b 100644
--- a/src/hydra-eval-jobs/hydra-eval-jobs.cc
+++ b/src/hydra-eval-jobs/hydra-eval-jobs.cc
@@ -89,7 +89,7 @@ struct MyArgs : MixEvalArgs, MixCommonArgs, RootArgs
 
 static MyArgs myArgs;
 
-static std::string queryMetaStrings(EvalState & state, DrvInfo & drv, const std::string & name, const std::string & subAttribute)
+static std::string queryMetaStrings(EvalState & state, PackageInfo & drv, const std::string & name, const std::string & subAttribute)
 {
     Strings res;
     std::function<void(Value & v)> rec;
@@ -181,7 +181,7 @@ static void worker(
                 // CA derivations do not have static output paths, so we
                 // have to defensively not query output paths in case we
                 // encounter one.
-                DrvInfo::Outputs outputs = drv->queryOutputs(
+                PackageInfo::Outputs outputs = drv->queryOutputs(
                     !experimentalFeatureSettings.isEnabled(Xp::CaDerivations));
 
                 if (drv->querySystem() == "unknown")
diff --git a/src/hydra-queue-runner/build-remote.cc b/src/hydra-queue-runner/build-remote.cc
index f81a5a0a..ad510e1b 100644
--- a/src/hydra-queue-runner/build-remote.cc
+++ b/src/hydra-queue-runner/build-remote.cc
@@ -8,6 +8,7 @@
 #include "build-result.hh"
 #include "path.hh"
 #include "serve-protocol.hh"
+#include "serve-protocol-impl.hh"
 #include "state.hh"
 #include "current-process.hh"
 #include "processes.hh"
@@ -20,12 +21,6 @@
 
 using namespace nix;
 
-
-static void append(Strings & dst, const Strings & src)
-{
-    dst.insert(dst.end(), src.begin(), src.end());
-}
-
 namespace nix::build_remote {
 
 static Strings extraStoreArgs(std::string & machine)
@@ -48,58 +43,20 @@ static Strings extraStoreArgs(std::string & machine)
     return result;
 }
 
-static void openConnection(::Machine::ptr machine, Path tmpDir, int stderrFD, SSHMaster::Connection & child)
+static std::unique_ptr<SSHMaster::Connection> openConnection(
+    ::Machine::ptr machine, SSHMaster & master)
 {
-    std::string pgmName;
-    Pipe to, from;
-    to.create();
-    from.create();
-
-    Strings argv;
+    Strings command = {"nix-store", "--serve", "--write"};
     if (machine->isLocalhost()) {
-        pgmName = "nix-store";
-        argv = {"nix-store", "--builders", "", "--serve", "--write"};
+        command.push_back("--builders");
+        command.push_back("");
     } else {
-        pgmName = "ssh";
-        auto sshName = machine->sshName;
-        Strings extraArgs = extraStoreArgs(sshName);
-        argv = {"ssh", sshName};
-        if (machine->sshKey != "") append(argv, {"-i", machine->sshKey});
-        if (machine->sshPublicHostKey != "") {
-            Path fileName = tmpDir + "/host-key";
-            auto p = machine->sshName.find("@");
-            std::string host = p != std::string::npos ? std::string(machine->sshName, p + 1) : machine->sshName;
-            writeFile(fileName, host + " " + machine->sshPublicHostKey + "\n");
-            append(argv, {"-oUserKnownHostsFile=" + fileName});
-        }
-        append(argv,
-            { "-x", "-a", "-oBatchMode=yes", "-oConnectTimeout=60", "-oTCPKeepAlive=yes"
-            , "--", "nix-store", "--serve", "--write" });
-        append(argv, extraArgs);
+        command.splice(command.end(), extraStoreArgs(machine->sshName));
     }
 
-    child.sshPid = startProcess([&]() {
-        restoreProcessContext();
-
-        if (dup2(to.readSide.get(), STDIN_FILENO) == -1)
-            throw SysError("cannot dup input pipe to stdin");
-
-        if (dup2(from.writeSide.get(), STDOUT_FILENO) == -1)
-            throw SysError("cannot dup output pipe to stdout");
-
-        if (dup2(stderrFD, STDERR_FILENO) == -1)
-            throw SysError("cannot dup stderr");
-
-        execvp(argv.front().c_str(), (char * *) stringsToCharPtrs(argv).data()); // FIXME: remove cast
-
-        throw SysError("cannot start %s", pgmName);
+    return master.startCommand(std::move(command), {
+        "-a", "-oBatchMode=yes", "-oConnectTimeout=60", "-oTCPKeepAlive=yes"
     });
-
-    to.readSide = -1;
-    from.writeSide = -1;
-
-    child.in = to.writeSide.release();
-    child.out = from.readSide.release();
 }
 
 
@@ -117,13 +74,10 @@ static void copyClosureTo(
        garbage-collect paths that are already there. Optionally, ask
        the remote host to substitute missing paths. */
     // FIXME: substitute output pollutes our build log
-    conn.to << ServeProto::Command::QueryValidPaths << 1 << useSubstitutes;
-    ServeProto::write(destStore, conn, closure);
-    conn.to.flush();
-
     /* Get back the set of paths that are already valid on the remote
        host. */
-    auto present = ServeProto::Serialise<StorePathSet>::read(destStore, conn);
+    auto present = conn.queryValidPaths(
+        destStore, true, closure, useSubstitutes);
 
     if (present.size() == closure.size()) return;
 
@@ -148,7 +102,7 @@ static void copyClosureTo(
 
 
 // FIXME: use Store::topoSortPaths().
-static StorePaths reverseTopoSortPaths(const std::map<StorePath, ValidPathInfo> & paths)
+static StorePaths reverseTopoSortPaths(const std::map<StorePath, UnkeyedValidPathInfo> & paths)
 {
     StorePaths sorted;
     StorePathSet visited;
@@ -189,28 +143,6 @@ static std::pair<Path, AutoCloseFD> openLogFile(const std::string & logDir, cons
     return {std::move(logFile), std::move(logFD)};
 }
 
-/**
- * @param conn is not fully initialized; it is this functions job to set
- * the `remoteVersion` field after the handshake is completed.
- * Therefore, no `ServeProto::Serialize` functions can be used until
- * that field is set.
- */
-static void handshake(::Machine::Connection & conn, unsigned int repeats)
-{
-    conn.to << SERVE_MAGIC_1 << 0x206;
-    conn.to.flush();
-
-    unsigned int magic = readInt(conn.from);
-    if (magic != SERVE_MAGIC_2)
-        throw Error("protocol mismatch with ‘nix-store --serve’ on ‘%1%’", conn.machine->sshName);
-    conn.remoteVersion = readInt(conn.from);
-    // Now `conn` is initialized.
-    if (GET_PROTOCOL_MAJOR(conn.remoteVersion) != 0x200)
-        throw Error("unsupported ‘nix-store --serve’ protocol version on ‘%1%’", conn.machine->sshName);
-    if (GET_PROTOCOL_MINOR(conn.remoteVersion) < 3 && repeats > 0)
-        throw Error("machine ‘%1%’ does not support repeating a build; please upgrade it to Nix 1.12", conn.machine->sshName);
-}
-
 static BasicDerivation sendInputs(
     State & state,
     Step & step,
@@ -281,21 +213,11 @@ static BuildResult performBuild(
     Store & localStore,
     StorePath drvPath,
     const BasicDerivation & drv,
-    const State::BuildOptions & options,
+    const ServeProto::BuildOptions & options,
     counter & nrStepsBuilding
 )
 {
-    conn.to << ServeProto::Command::BuildDerivation << localStore.printStorePath(drvPath);
-    writeDerivation(conn.to, localStore, drv);
-    conn.to << options.maxSilentTime << options.buildTimeout;
-    if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 2)
-        conn.to << options.maxLogSize;
-    if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 3) {
-        conn.to
-            << options.repeats // == build-repeat
-            << options.enforceDeterminism;
-    }
-    conn.to.flush();
+    conn.putBuildDerivationRequest(localStore, drvPath, drv, options);
 
     BuildResult result;
 
@@ -345,7 +267,7 @@ static BuildResult performBuild(
     return result;
 }
 
-static std::map<StorePath, ValidPathInfo> queryPathInfos(
+static std::map<StorePath, UnkeyedValidPathInfo> queryPathInfos(
     ::Machine::Connection & conn,
     Store & localStore,
     StorePathSet & outputs,
@@ -354,30 +276,18 @@ static std::map<StorePath, ValidPathInfo> queryPathInfos(
 {
 
     /* Get info about each output path. */
-    std::map<StorePath, ValidPathInfo> infos;
+    std::map<StorePath, UnkeyedValidPathInfo> infos;
     conn.to << ServeProto::Command::QueryPathInfos;
     ServeProto::write(localStore, conn, outputs);
     conn.to.flush();
     while (true) {
         auto storePathS = readString(conn.from);
         if (storePathS == "") break;
-        auto deriver = readString(conn.from); // deriver
-        auto references = ServeProto::Serialise<StorePathSet>::read(localStore, conn);
-        readLongLong(conn.from); // download size
-        auto narSize = readLongLong(conn.from);
-        auto narHash = Hash::parseAny(readString(conn.from), htSHA256);
-        auto ca = ContentAddress::parseOpt(readString(conn.from));
-        readStrings<StringSet>(conn.from); // sigs
-        ValidPathInfo info(localStore.parseStorePath(storePathS), narHash);
-        assert(outputs.count(info.path));
-        info.references = references;
-        info.narSize = narSize;
+
+        auto storePath = localStore.parseStorePath(storePathS);
+        auto info = ServeProto::Serialise<UnkeyedValidPathInfo>::read(localStore, conn);
         totalNarSize += info.narSize;
-        info.narHash = narHash;
-        info.ca = ca;
-        if (deriver != "")
-            info.deriver = localStore.parseStorePath(deriver);
-        infos.insert_or_assign(info.path, info);
+        infos.insert_or_assign(std::move(storePath), std::move(info));
     }
 
     return infos;
@@ -418,14 +328,16 @@ static void copyPathsFromRemote(
     NarMemberDatas & narMembers,
     Store & localStore,
     Store & destStore,
-    const std::map<StorePath, ValidPathInfo> & infos
+    const std::map<StorePath, UnkeyedValidPathInfo> & infos
 )
 {
       auto pathsSorted = reverseTopoSortPaths(infos);
 
       for (auto & path : pathsSorted) {
           auto & info = infos.find(path)->second;
-          copyPathFromRemote(conn, narMembers, localStore, destStore, info);
+          copyPathFromRemote(
+              conn, narMembers, localStore, destStore,
+              ValidPathInfo { path, info });
       }
 
 }
@@ -492,7 +404,7 @@ void RemoteResult::updateWithBuildResult(const nix::BuildResult & buildResult)
 
 void State::buildRemote(ref<Store> destStore,
     ::Machine::ptr machine, Step::ptr step,
-    const BuildOptions & buildOptions,
+    const ServeProto::BuildOptions & buildOptions,
     RemoteResult & result, std::shared_ptr<ActiveStep> activeStep,
     std::function<void(StepState)> updateStep,
     NarMemberDatas & narMembers)
@@ -503,21 +415,26 @@ void State::buildRemote(ref<Store> destStore,
     AutoDelete logFileDel(logFile, false);
     result.logFile = logFile;
 
-    nix::Path tmpDir = createTempDir();
-    AutoDelete tmpDirDel(tmpDir, true);
-
     try {
 
         updateStep(ssConnecting);
 
+        SSHMaster master {
+            machine->sshName,
+            machine->sshKey,
+            machine->sshPublicHostKey,
+            false, // no SSH master yet
+            false, // no compression yet
+            logFD.get(),
+        };
+
         // FIXME: rewrite to use Store.
-        SSHMaster::Connection child;
-        build_remote::openConnection(machine, tmpDir, logFD.get(), child);
+        auto child = build_remote::openConnection(machine, master);
 
         {
             auto activeStepState(activeStep->state_.lock());
             if (activeStepState->cancelled) throw Error("step cancelled");
-            activeStepState->pid = child.sshPid;
+            activeStepState->pid = child->sshPid;
         }
 
         Finally clearPid([&]() {
@@ -533,9 +450,13 @@ void State::buildRemote(ref<Store> destStore,
         });
 
         ::Machine::Connection conn {
-            .from = child.out.get(),
-            .to = child.in.get(),
-            .machine = machine,
+            {
+                .to = child->in.get(),
+                .from = child->out.get(),
+                /* Handshake. */
+                .remoteVersion = 0xdadbeef, // FIXME avoid dummy initialize
+            },
+            /*.machine =*/ machine,
         };
 
         Finally updateStats([&]() {
@@ -543,14 +464,26 @@ void State::buildRemote(ref<Store> destStore,
             bytesSent += conn.to.written;
         });
 
+        constexpr ServeProto::Version our_version = 0x206;
+
         try {
-          build_remote::handshake(conn, buildOptions.repeats);
+            conn.remoteVersion = decltype(conn)::handshake(
+                conn.to,
+                conn.from,
+                our_version,
+                machine->sshName);
         } catch (EndOfFile & e) {
-            child.sshPid.wait();
+            child->sshPid.wait();
             std::string s = chomp(readFile(result.logFile));
             throw Error("cannot connect to ‘%1%’: %2%", machine->sshName, s);
         }
 
+        // Do not attempt to speak a newer version of the protocol.
+        //
+        // Per https://github.com/NixOS/nix/issues/9584 should be handled as
+        // part of `handshake` in upstream nix.
+        conn.remoteVersion = std::min(conn.remoteVersion, our_version);
+
         {
             auto info(machine->state->connectInfo.lock());
             info->consecutiveFailures = 0;
@@ -653,8 +586,8 @@ void State::buildRemote(ref<Store> destStore,
         }
 
         /* Shut down the connection. */
-        child.in = -1;
-        child.sshPid.wait();
+        child->in = -1;
+        child->sshPid.wait();
 
     } catch (Error & e) {
         /* Disable this machine until a certain period of time has
diff --git a/src/hydra-queue-runner/builder.cc b/src/hydra-queue-runner/builder.cc
index 0ef77068..5269febd 100644
--- a/src/hydra-queue-runner/builder.cc
+++ b/src/hydra-queue-runner/builder.cc
@@ -98,10 +98,13 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
        it). */
     BuildID buildId;
     std::optional<StorePath> buildDrvPath;
-    BuildOptions buildOptions;
-    buildOptions.repeats = step->isDeterministic ? 1 : 0;
-    buildOptions.maxLogSize = maxLogSize;
-    buildOptions.enforceDeterminism = step->isDeterministic;
+    // Other fields set below
+    nix::ServeProto::BuildOptions buildOptions {
+        .maxLogSize = maxLogSize,
+        .nrRepeats = step->isDeterministic ? 1u : 0u,
+        .enforceDeterminism = step->isDeterministic,
+        .keepFailed = false,
+    };
 
     auto conn(dbPool.get());
 
@@ -136,7 +139,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
             {
                 auto i = jobsetRepeats.find(std::make_pair(build2->projectName, build2->jobsetName));
                 if (i != jobsetRepeats.end())
-                    buildOptions.repeats = std::max(buildOptions.repeats, i->second);
+                    buildOptions.nrRepeats = std::max(buildOptions.nrRepeats, i->second);
             }
         }
         if (!build) build = *dependents.begin();
@@ -147,7 +150,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
         buildOptions.buildTimeout = build->buildTimeout;
 
         printInfo("performing step ‘%s’ %d times on ‘%s’ (needed by build %d and %d others)",
-            localStore->printStorePath(step->drvPath), buildOptions.repeats + 1, machine->sshName, buildId, (dependents.size() - 1));
+            localStore->printStorePath(step->drvPath), buildOptions.nrRepeats + 1, machine->sshName, buildId, (dependents.size() - 1));
     }
 
     if (!buildOneDone)
diff --git a/src/hydra-queue-runner/dispatcher.cc b/src/hydra-queue-runner/dispatcher.cc
index f5d9d3e3..6d738ded 100644
--- a/src/hydra-queue-runner/dispatcher.cc
+++ b/src/hydra-queue-runner/dispatcher.cc
@@ -231,11 +231,11 @@ system_time State::doDispatch()
         sort(machinesSorted.begin(), machinesSorted.end(),
             [](const MachineInfo & a, const MachineInfo & b) -> bool
             {
-                float ta = std::round(a.currentJobs / a.machine->speedFactorFloat);
-                float tb = std::round(b.currentJobs / b.machine->speedFactorFloat);
+                float ta = std::round(a.currentJobs / a.machine->speedFactor);
+                float tb = std::round(b.currentJobs / b.machine->speedFactor);
                 return
                     ta != tb ? ta < tb :
-                    a.machine->speedFactorFloat != b.machine->speedFactorFloat ? a.machine->speedFactorFloat > b.machine->speedFactorFloat :
+                    a.machine->speedFactor != b.machine->speedFactor ? a.machine->speedFactor > b.machine->speedFactor :
                     a.currentJobs > b.currentJobs;
             });
 
diff --git a/src/hydra-queue-runner/hydra-queue-runner.cc b/src/hydra-queue-runner/hydra-queue-runner.cc
index 0f4e759c..0ee710cb 100644
--- a/src/hydra-queue-runner/hydra-queue-runner.cc
+++ b/src/hydra-queue-runner/hydra-queue-runner.cc
@@ -155,16 +155,16 @@ void State::parseMachines(const std::string & contents)
         auto machine = std::make_shared<::Machine>(nix::Machine {
             // `storeUri`, not yet used
             "",
-            // `systemTypes`, not yet used
-            {},
+            // `systemTypes`
+            tokenizeString<StringSet>(tokens[1], ","),
             // `sshKey`
             tokens[2] == "-" ? "" : tokens[2],
             // `maxJobs`
             tokens[3] != ""
                 ? string2Int<MaxJobs>(tokens[3]).value()
                 : 1,
-            // `speedFactor`, not yet used
-            1,
+            // `speedFactor`
+            atof(tokens[4].c_str()),
             // `supportedFeatures`
             std::move(supportedFeatures),
             // `mandatoryFeatures`
@@ -176,8 +176,6 @@ void State::parseMachines(const std::string & contents)
         });
 
         machine->sshName = tokens[0];
-        machine->systemTypesSet = tokenizeString<StringSet>(tokens[1], ",");
-        machine->speedFactorFloat = atof(tokens[4].c_str());
 
         /* Re-use the State object of the previous machine with the
            same name. */
@@ -641,7 +639,7 @@ void State::dumpStatus(Connection & conn)
 
                 json machine = {
                     {"enabled",  m->enabled},
-                    {"systemTypes", m->systemTypesSet},
+                    {"systemTypes", m->systemTypes},
                     {"supportedFeatures", m->supportedFeatures},
                     {"mandatoryFeatures", m->mandatoryFeatures},
                     {"nrStepsDone", s->nrStepsDone.load()},
diff --git a/src/hydra-queue-runner/nar-extractor.cc b/src/hydra-queue-runner/nar-extractor.cc
index 3c6857bf..61299ecd 100644
--- a/src/hydra-queue-runner/nar-extractor.cc
+++ b/src/hydra-queue-runner/nar-extractor.cc
@@ -6,7 +6,46 @@
 
 using namespace nix;
 
-struct Extractor : ParseSink
+
+struct NarMemberConstructor : CreateRegularFileSink
+{
+    NarMemberData & curMember;
+
+    HashSink hashSink = HashSink { HashAlgorithm::SHA256 };
+
+    std::optional<uint64_t> expectedSize;
+
+    NarMemberConstructor(NarMemberData & curMember)
+        : curMember(curMember)
+    { }
+
+    void isExecutable() override
+    {
+    }
+
+    void preallocateContents(uint64_t size) override
+    {
+        expectedSize = size;
+    }
+
+    void operator () (std::string_view data) override
+    {
+        assert(expectedSize);
+        *curMember.fileSize += data.size();
+        hashSink(data);
+        if (curMember.contents) {
+            curMember.contents->append(data);
+        }
+        assert(curMember.fileSize <= expectedSize);
+        if (curMember.fileSize == expectedSize) {
+            auto [hash, len] = hashSink.finish();
+            assert(curMember.fileSize == len);
+            curMember.sha256 = hash;
+        }
+    }
+};
+
+struct Extractor : FileSystemObjectSink
 {
     std::unordered_set<Path> filesToKeep {
         "/nix-support/hydra-build-products",
@@ -15,7 +54,6 @@ struct Extractor : ParseSink
     };
 
     NarMemberDatas & members;
-    NarMemberData * curMember = nullptr;
     Path prefix;
 
     Extractor(NarMemberDatas & members, const Path & prefix)
@@ -27,53 +65,22 @@ struct Extractor : ParseSink
         members.insert_or_assign(prefix + path, NarMemberData { .type = SourceAccessor::Type::tDirectory });
     }
 
-    void createRegularFile(const Path & path) override
+    void createRegularFile(const Path & path, std::function<void(CreateRegularFileSink &)> func) override
     {
-        curMember = &members.insert_or_assign(prefix + path, NarMemberData {
-            .type = SourceAccessor::Type::tRegular,
-            .fileSize = 0,
-            .contents = filesToKeep.count(path) ? std::optional("") : std::nullopt,
-        }).first->second;
-    }
-
-    std::optional<uint64_t> expectedSize;
-    std::unique_ptr<HashSink> hashSink;
-
-    void preallocateContents(uint64_t size) override
-    {
-        expectedSize = size;
-        hashSink = std::make_unique<HashSink>(htSHA256);
-    }
-
-    void receiveContents(std::string_view data) override
-    {
-        assert(expectedSize);
-        assert(curMember);
-        assert(hashSink);
-        *curMember->fileSize += data.size();
-        (*hashSink)(data);
-        if (curMember->contents) {
-            curMember->contents->append(data);
-        }
-        assert(curMember->fileSize <= expectedSize);
-        if (curMember->fileSize == expectedSize) {
-            auto [hash, len] = hashSink->finish();
-            assert(curMember->fileSize == len);
-            curMember->sha256 = hash;
-            hashSink.reset();
-        }
+        NarMemberConstructor nmc {
+            members.insert_or_assign(prefix + path, NarMemberData {
+                .type = SourceAccessor::Type::tRegular,
+                .fileSize = 0,
+                .contents = filesToKeep.count(path) ? std::optional("") : std::nullopt,
+            }).first->second,
+        };
+        func(nmc);
     }
 
     void createSymlink(const Path & path, const std::string & target) override
     {
         members.insert_or_assign(prefix + path, NarMemberData { .type = SourceAccessor::Type::tSymlink });
     }
-
-    void isExecutable() override
-    { }
-
-    void closeRegularFile() override
-    { }
 };
 
 
diff --git a/src/hydra-queue-runner/queue-monitor.cc b/src/hydra-queue-runner/queue-monitor.cc
index d7214c43..203f9f1d 100644
--- a/src/hydra-queue-runner/queue-monitor.cc
+++ b/src/hydra-queue-runner/queue-monitor.cc
@@ -696,7 +696,7 @@ BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store>
                 product.fileSize = row[2].as<off_t>();
             }
             if (!row[3].is_null())
-                product.sha256hash = Hash::parseAny(row[3].as<std::string>(), htSHA256);
+                product.sha256hash = Hash::parseAny(row[3].as<std::string>(), HashAlgorithm::SHA256);
             if (!row[4].is_null())
                 product.path = row[4].as<std::string>();
             product.name = row[5].as<std::string>();
diff --git a/src/hydra-queue-runner/state.hh b/src/hydra-queue-runner/state.hh
index d1c78c68..cda238ae 100644
--- a/src/hydra-queue-runner/state.hh
+++ b/src/hydra-queue-runner/state.hh
@@ -22,6 +22,7 @@
 #include "sync.hh"
 #include "nar-extractor.hh"
 #include "serve-protocol.hh"
+#include "serve-protocol-impl.hh"
 #include "machines.hh"
 
 
@@ -243,14 +244,6 @@ struct Machine : nix::Machine
        we are not yet used to, but once we are, we don't need this. */
     std::string sshName;
 
-    /* TODO Get rid once `nix::Machine::systemTypes` is a set not
-       vector. */
-    std::set<std::string> systemTypesSet;
-
-    /* TODO Get rid once `nix::Machine::systemTypes` is a `float` not
-       an `int`. */
-    float speedFactorFloat = 1.0;
-
     struct State {
         typedef std::shared_ptr<State> ptr;
         counter currentJobs{0};
@@ -277,7 +270,7 @@ struct Machine : nix::Machine
     {
         /* Check that this machine is of the type required by the
            step. */
-        if (!systemTypesSet.count(step->drv->platform == "builtin" ? nix::settings.thisSystem : step->drv->platform))
+        if (!systemTypes.count(step->drv->platform == "builtin" ? nix::settings.thisSystem : step->drv->platform))
             return false;
 
         /* Check that the step requires all mandatory features of this
@@ -307,29 +300,9 @@ struct Machine : nix::Machine
     }
 
     // A connection to a machine
-    struct Connection {
-        nix::FdSource from;
-        nix::FdSink to;
-        nix::ServeProto::Version remoteVersion;
-
+    struct Connection : nix::ServeProto::BasicClientConnection {
         // Backpointer to the machine
         ptr machine;
-
-        operator nix::ServeProto::ReadConn ()
-        {
-            return {
-                .from = from,
-                .version = remoteVersion,
-            };
-        }
-
-        operator nix::ServeProto::WriteConn ()
-        {
-            return {
-                .to = to,
-                .version = remoteVersion,
-            };
-        }
     };
 };
 
@@ -464,7 +437,7 @@ private:
 
     /* How often the build steps of a jobset should be repeated in
        order to detect non-determinism. */
-    std::map<std::pair<std::string, std::string>, unsigned int> jobsetRepeats;
+    std::map<std::pair<std::string, std::string>, size_t> jobsetRepeats;
 
     bool uploadLogsToBinaryCache;
 
@@ -493,12 +466,6 @@ private:
 public:
     State(std::optional<std::string> metricsAddrOpt);
 
-    struct BuildOptions {
-        unsigned int maxSilentTime, buildTimeout, repeats;
-        size_t maxLogSize;
-        bool enforceDeterminism;
-    };
-
 private:
 
     nix::MaintainCount<counter> startDbUpdate();
@@ -583,7 +550,7 @@ private:
 
     void buildRemote(nix::ref<nix::Store> destStore,
         Machine::ptr machine, Step::ptr step,
-        const BuildOptions & buildOptions,
+        const nix::ServeProto::BuildOptions & buildOptions,
         RemoteResult & result, std::shared_ptr<ActiveStep> activeStep,
         std::function<void(StepState)> updateStep,
         NarMemberDatas & narMembers);