Merge branch 'ca/track-drvoutput-dependencies-2-le-retour' into ca/recursively-substitute-realisations

src/libstore/build/derivation-goal.cc

@@ -927,6 +927,7 @@ void DerivationGoal::resolvedFinished() {
             auto newRealisation = *realisation;
             newRealisation.id = DrvOutput{initialOutputs.at(wantedOutput).outputHash, wantedOutput};
             newRealisation.signatures.clear();
+            newRealisation.dependentRealisations = drvOutputReferences(worker.store, *drv, realisation->outPath);
             signRealisation(newRealisation);
             worker.store.registerDrvOutput(newRealisation);
         } else {

src/libstore/ca-specific-schema.sql

@@ -3,10 +3,19 @@
 -- is enabled
 
 create table if not exists Realisations (
+    id integer primary key autoincrement not null,
     drvPath text not null,
     outputName text not null, -- symbolic output id, usually "out"
     outputPath integer not null,
     signatures text, -- space-separated list
-    primary key (drvPath, outputName),
     foreign key (outputPath) references ValidPaths(id) on delete cascade
 );
+
+create index if not exists IndexRealisations on Realisations(drvPath, outputName);
+
+create table if not exists RealisationsRefs (
+    referrer integer not null,
+    realisationReference integer,
+    foreign key (referrer) references Realisations(id) on delete cascade,
+    foreign key (realisationReference) references Realisations(id) on delete restrict
+);

src/libstore/daemon.cc

@@ -885,10 +885,15 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
 
     case wopRegisterDrvOutput: {
         logger->startWork();
+        if (GET_PROTOCOL_MINOR(clientVersion) < 31) {
             auto outputId = DrvOutput::parse(readString(from));
             auto outputPath = StorePath(readString(from));
             store->registerDrvOutput(Realisation{
                 .id = outputId, .outPath = outputPath});
+        } else {
+            auto realisation = worker_proto::read(*store, from, Phantom<Realisation>());
+            store->registerDrvOutput(realisation);
+        }
         logger->stopWork();
         break;
     }
@@ -898,9 +903,15 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         auto outputId = DrvOutput::parse(readString(from));
         auto info = store->queryRealisation(outputId);
         logger->stopWork();
+        if (GET_PROTOCOL_MINOR(clientVersion) < 31) {
             std::set<StorePath> outPaths;
             if (info) outPaths.insert(info->outPath);
             worker_proto::write(*store, to, outPaths);
+        } else {
+            std::set<Realisation> realisations;
+            if (info) realisations.insert(*info);
+            worker_proto::write(*store, to, realisations);
+        }
         break;
     }
 

src/libstore/local-store.cc

@@ -59,6 +59,8 @@ struct LocalStore::State::Stmts {
     SQLiteStmt QueryAllRealisedOutputs;
     SQLiteStmt QueryPathFromHashPart;
     SQLiteStmt QueryValidPaths;
+    SQLiteStmt QueryRealisationRealisationReferences;
+    SQLiteStmt AddRealisationRealisationReference;
 };
 
 int getSchema(Path schemaPath)
@@ -76,7 +78,7 @@ int getSchema(Path schemaPath)
 
 void migrateCASchema(SQLite& db, Path schemaPath, AutoCloseFD& lockFd)
 {
-    const int nixCASchemaVersion = 1;
+    const int nixCASchemaVersion = 2;
     int curCASchema = getSchema(schemaPath);
     if (curCASchema != nixCASchemaVersion) {
         if (curCASchema > nixCASchemaVersion) {
@@ -94,7 +96,39 @@ void migrateCASchema(SQLite& db, Path schemaPath, AutoCloseFD& lockFd)
               #include "ca-specific-schema.sql.gen.hh"
                 ;
             db.exec(schema);
+            curCASchema = nixCASchemaVersion;
         }
+
+        if (curCASchema < 2) {
+            SQLiteTxn txn(db);
+            // Ugly little sql dance to add a new `id` column and make it the primary key
+            db.exec(R"(
+                create table Realisations2 (
+                    id integer primary key autoincrement not null,
+                    drvPath text not null,
+                    outputName text not null, -- symbolic output id, usually "out"
+                    outputPath integer not null,
+                    signatures text, -- space-separated list
+                    foreign key (outputPath) references ValidPaths(id) on delete cascade
+                );
+                insert into Realisations2 (drvPath, outputName, outputPath, signatures)
+                    select drvPath, outputName, outputPath, signatures from Realisations;
+                drop table Realisations;
+                alter table Realisations2 rename to Realisations;
+            )");
+            db.exec(R"(
+                create index if not exists IndexRealisations on Realisations(drvPath, outputName);
+
+                create table if not exists RealisationsRefs (
+                    referrer integer not null,
+                    realisationReference integer,
+                    foreign key (referrer) references Realisations(id) on delete cascade,
+                    foreign key (realisationReference) references Realisations(id) on delete restrict
+                );
+            )");
+            txn.commit();
+        }
+
         writeFile(schemaPath, fmt("%d", nixCASchemaVersion));
         lockFile(lockFd.get(), ltRead, true);
     }
@@ -316,7 +350,7 @@ LocalStore::LocalStore(const Params & params)
             )");
         state->stmts->QueryRealisedOutput.create(state->db,
             R"(
-                select Output.path, Realisations.signatures from Realisations
+                select Realisations.id, Output.path, Realisations.signatures from Realisations
                     inner join ValidPaths as Output on Output.id = Realisations.outputPath
                     where drvPath = ? and outputName = ?
                     ;
@@ -328,6 +362,19 @@ LocalStore::LocalStore(const Params & params)
                     where drvPath = ?
                     ;
             )");
+        state->stmts->QueryRealisationRealisationReferences.create(state->db,
+            R"(
+                select drvPath, outputName from Realisations
+                    join RealisationsRefs on realisationReference = Realisations.id
+                    where referrer = ?;
+            )");
+        state->stmts->AddRealisationRealisationReference.create(state->db,
+            R"(
+                insert or replace into RealisationsRefs (referrer, realisationReference)
+                values (
+                    ?,
+                    (select id from Realisations where drvPath = ? and outputName = ?));
+            )");
     }
 }
 
@@ -664,14 +711,18 @@ void LocalStore::registerDrvOutput(const Realisation & info, CheckSigsFlag check
 void LocalStore::registerDrvOutput(const Realisation & info)
 {
     settings.requireExperimentalFeature("ca-derivations");
-    auto state(_state.lock());
     retrySQLite<void>([&]() {
-        state->stmts->RegisterRealisedOutput.use()
+        auto state(_state.lock());
-            (info.id.strHash())
+        state->stmts->RegisterRealisedOutput
-            (info.id.outputName)
+            .use()(info.id.strHash())(info.id.outputName)(printStorePath(
-            (printStorePath(info.outPath))
+                info.outPath))(concatStringsSep(" ", info.signatures))
-            (concatStringsSep(" ", info.signatures))
             .exec();
+        uint64_t myId = state->db.getLastInsertedRowId();
+        for (auto & [outputId, _] : info.dependentRealisations) {
+            state->stmts->AddRealisationRealisationReference
+                .use()(myId)(outputId.strHash())(outputId.outputName)
+                .exec();
+        }
     });
 }
 
@@ -1670,14 +1721,36 @@ std::optional<const Realisation> LocalStore::queryRealisation(
     typedef std::optional<const Realisation> Ret;
     return retrySQLite<Ret>([&]() -> Ret {
         auto state(_state.lock());
-        auto use(state->stmts->QueryRealisedOutput.use()(id.strHash())(
+        auto useQueryRealisedOutput(state->stmts->QueryRealisedOutput.use()(
-            id.outputName));
+            id.strHash())(id.outputName));
-        if (!use.next())
+        if (!useQueryRealisedOutput.next())
             return std::nullopt;
-        auto outputPath = parseStorePath(use.getStr(0));
+        auto realisationDbId = useQueryRealisedOutput.getInt(0);
-        auto signatures = tokenizeString<StringSet>(use.getStr(1));
+        auto outputPath = parseStorePath(useQueryRealisedOutput.getStr(1));
+        auto signatures =
+            tokenizeString<StringSet>(useQueryRealisedOutput.getStr(2));
+
+        std::map<DrvOutput, StorePath> dependentRealisations;
+        auto useRealisationRefs(
+            state->stmts->QueryRealisationRealisationReferences.use()(
+                realisationDbId));
+        while (useRealisationRefs.next()) {
+            auto depHash = useRealisationRefs.getStr(0);
+            auto depOutputName = useRealisationRefs.getStr(1);
+            auto useQueryRealisedOutput(
+                state->stmts->QueryRealisedOutput.use()(depHash)(depOutputName));
+            assert(useQueryRealisedOutput.next());
+            auto outputPath = parseStorePath(useQueryRealisedOutput.getStr(1));
+            auto depId = DrvOutput { Hash::parseAnyPrefixed(depHash), depOutputName };
+            dependentRealisations.insert({depId, outputPath});
+        }
+
         return Ret{Realisation{
-            .id = id, .outPath = outputPath, .signatures = signatures}};
+            .id = id,
+            .outPath = outputPath,
+            .signatures = signatures,
+            .dependentRealisations = dependentRealisations,
+        }};
     });
 }
 }  // namespace nix

src/libstore/misc.cc

@@ -254,5 +254,44 @@ StorePaths Store::topoSortPaths(const StorePathSet & paths)
         }});
 }
 
+std::map<DrvOutput, StorePath> drvOutputReferences(
+    const std::set<Realisation> inputRealisations,
+    const StorePathSet pathReferences)
+{
+    std::map<DrvOutput, StorePath> res;
 
+    for (const auto & input : inputRealisations) {
+        if (pathReferences.count(input.outPath)) {
+            res.insert({input.id, input.outPath});
+        }
+    }
+
+    return res;
+}
+
+std::map<DrvOutput, StorePath> drvOutputReferences(
+    Store & store,
+    const Derivation & drv,
+    const StorePath & outputPath)
+{
+    std::set<Realisation> inputRealisations;
+
+    for (const auto& [inputDrv, outputNames] : drv.inputDrvs) {
+        auto outputHashes =
+            staticOutputHashes(store, store.readDerivation(inputDrv));
+        for (const auto& outputName : outputNames) {
+            auto thisRealisation = store.queryRealisation(
+                DrvOutput{outputHashes.at(outputName), outputName});
+            if (!thisRealisation)
+                throw Error(
+                    "output '%s' of derivation '%s' isn’t built", outputName,
+                    store.printStorePath(inputDrv));
+            inputRealisations.insert(*thisRealisation);
+        }
+    }
+
+    auto info = store.queryPathInfo(outputPath);
+
+    return drvOutputReferences(Realisation::closure(store, inputRealisations), info->references);
+}
 }

src/libstore/realisation.cc

@@ -1,5 +1,6 @@
 #include "realisation.hh"
 #include "store-api.hh"
+#include "closure.hh"
 #include <nlohmann/json.hpp>
 
 namespace nix {
@@ -21,11 +22,52 @@ std::string DrvOutput::to_string() const {
     return strHash() + "!" + outputName;
 }
 
+std::set<Realisation> Realisation::closure(Store & store, std::set<Realisation> startOutputs)
+{
+    std::set<Realisation> res;
+    Realisation::closure(store, startOutputs, res);
+    return res;
+}
+
+void Realisation::closure(Store & store, std::set<Realisation> startOutputs, std::set<Realisation> & res)
+{
+    auto getDeps = [&](const Realisation& current) -> std::set<Realisation> {
+        std::set<Realisation> res;
+        for (auto& [currentDep, _] : current.dependentRealisations) {
+            if (auto currentRealisation = store.queryRealisation(currentDep))
+                res.insert(*currentRealisation);
+            else
+                throw Error(
+                    "Unrealised derivation '%s'", currentDep.to_string());
+        }
+        return res;
+    };
+
+    computeClosure<Realisation>(
+        startOutputs, res,
+        [&](const Realisation& current,
+            std::function<void(std::promise<std::set<Realisation>>&)>
+                processEdges) {
+            std::promise<std::set<Realisation>> promise;
+            try {
+                auto res = getDeps(current);
+                promise.set_value(res);
+            } catch (...) {
+                promise.set_exception(std::current_exception());
+            }
+            return processEdges(promise);
+        });
+}
+
 nlohmann::json Realisation::toJSON() const {
+    auto jsonDependentRealisations = nlohmann::json::object();
+    for (auto & [depId, depOutPath] : dependentRealisations)
+        jsonDependentRealisations.emplace(depId.to_string(), depOutPath.to_string());
     return nlohmann::json{
         {"id", id.to_string()},
         {"outPath", outPath.to_string()},
         {"signatures", signatures},
+        {"dependentRealisations", jsonDependentRealisations},
     };
 }
 
@@ -51,10 +93,16 @@ Realisation Realisation::fromJSON(
     if (auto signaturesIterator = json.find("signatures"); signaturesIterator != json.end())
         signatures.insert(signaturesIterator->begin(), signaturesIterator->end());
 
+    std::map <DrvOutput, StorePath> dependentRealisations;
+    if (auto jsonDependencies = json.find("dependentRealisations"); jsonDependencies != json.end())
+        for (auto & [jsonDepId, jsonDepOutPath] : jsonDependencies->get<std::map<std::string, std::string>>())
+            dependentRealisations.insert({DrvOutput::parse(jsonDepId), StorePath(jsonDepOutPath)});
+
     return Realisation{
         .id = DrvOutput::parse(getField("id")),
         .outPath = StorePath(getField("outPath")),
         .signatures = signatures,
+        .dependentRealisations = dependentRealisations,
     };
 }
 

src/libstore/realisation.hh

@@ -28,6 +28,14 @@ struct Realisation {
 
     StringSet signatures;
 
+    /**
+     * The realisations that are required for the current one to be valid.
+     *
+     * When importing this realisation, the store will first check that all its
+     * dependencies exist, and map to the correct output path
+     */
+    std::map<DrvOutput, StorePath> dependentRealisations;
+
     nlohmann::json toJSON() const;
     static Realisation fromJSON(const nlohmann::json& json, const std::string& whence);
 
@@ -36,6 +44,9 @@ struct Realisation {
     bool checkSignature(const PublicKeys & publicKeys, const std::string & sig) const;
     size_t checkSignatures(const PublicKeys & publicKeys) const;
 
+    static std::set<Realisation> closure(Store &, std::set<Realisation>);
+    static void closure(Store &, std::set<Realisation>, std::set<Realisation>& res);
+
     StorePath getPath() const { return outPath; }
 
     GENERATE_CMP(Realisation, me->id, me->outPath);

src/libstore/remote-store.cc

@@ -653,8 +653,12 @@ void RemoteStore::registerDrvOutput(const Realisation & info)
 {
     auto conn(getConnection());
     conn->to << wopRegisterDrvOutput;
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
         conn->to << info.id.to_string();
         conn->to << std::string(info.outPath.to_string());
+    } else {
+        worker_proto::write(*this, conn->to, info);
+    }
     conn.processStderr();
 }
 
@@ -664,10 +668,17 @@ std::optional<const Realisation> RemoteStore::queryRealisation(const DrvOutput &
     conn->to << wopQueryRealisation;
     conn->to << id.to_string();
     conn.processStderr();
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 31) {
         auto outPaths = worker_proto::read(*this, conn->from, Phantom<std::set<StorePath>>{});
         if (outPaths.empty())
             return std::nullopt;
         return {Realisation{.id = id, .outPath = *outPaths.begin()}};
+    } else {
+        auto realisations = worker_proto::read(*this, conn->from, Phantom<std::set<Realisation>>{});
+        if (realisations.empty())
+            return std::nullopt;
+        return *realisations.begin();
+    }
 }
 
 static void writeDerivedPaths(RemoteStore & store, ConnectionHandle & conn, const std::vector<DerivedPath> & reqs)

src/libstore/store-api.cc

@@ -780,20 +780,40 @@ std::map<StorePath, StorePath> copyPaths(ref<Store> srcStore, ref<Store> dstStor
     RepairFlag repair, CheckSigsFlag checkSigs, SubstituteFlag substitute)
 {
     StorePathSet storePaths;
-    std::set<Realisation> realisations;
+    std::set<Realisation> toplevelRealisations;
     for (auto & path : paths) {
         storePaths.insert(path.path());
         if (auto realisation = std::get_if<Realisation>(&path.raw)) {
             settings.requireExperimentalFeature("ca-derivations");
-            realisations.insert(*realisation);
+            toplevelRealisations.insert(*realisation);
         }
     }
     auto pathsMap = copyPaths(srcStore, dstStore, storePaths, repair, checkSigs, substitute);
+
+    ThreadPool pool;
+
     try {
-        for (auto & realisation : realisations) {
+        // Copy the realisation closure
-            dstStore->registerDrvOutput(realisation, checkSigs);
+        processGraph<Realisation>(
+            pool, Realisation::closure(*srcStore, toplevelRealisations),
+            [&](const Realisation& current) -> std::set<Realisation> {
+                std::set<Realisation> children;
+                for (const auto& [drvOutput, _] : current.dependentRealisations) {
+                    auto currentChild = srcStore->queryRealisation(drvOutput);
+                    if (!currentChild)
+                        throw Error(
+                            "Incomplete realisation closure: '%s' is a "
+                            "dependency "
+                            "of '%s' but isn’t registered",
+                            drvOutput.to_string(), current.id.to_string());
+                    children.insert(*currentChild);
                 }
-    } catch (MissingExperimentalFeature & e) {
+                return children;
+            },
+            [&](const Realisation& current) -> void {
+                dstStore->registerDrvOutput(current, checkSigs);
+            });
+    } catch (MissingExperimentalFeature& e) {
         // Don't fail if the remote doesn't support CA derivations is it might
         // not be within our control to change that, and we might still want
         // to at least copy the output paths.

src/libstore/store-api.hh

@@ -864,4 +864,9 @@ std::pair<std::string, Store::Params> splitUriAndParams(const std::string & uri)
 
 std::optional<ContentAddress> getDerivationCA(const BasicDerivation & drv);
 
+std::map<DrvOutput, StorePath> drvOutputReferences(
+    Store & store,
+    const Derivation & drv,
+    const StorePath & outputPath);
+
 }

src/libstore/worker-protocol.hh

@@ -9,7 +9,7 @@ namespace nix {
 #define WORKER_MAGIC_1 0x6e697863
 #define WORKER_MAGIC_2 0x6478696f
 
-#define PROTOCOL_VERSION (1 << 8 | 30)
+#define PROTOCOL_VERSION (1 << 8 | 31)
 #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)