ahuston-0/hydra
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

Fix libpqxx 7.10.1 API compatibility

Browse Source 
- Replace deprecated exec_params/exec_params0 calls with exec()
- Wrap all parameterized queries with pqxx::params{}
- Add .no_rows()/.one_row() to exec calls that don't return results
This commit is contained in:
Jörg Thalheim
2025-08-03 11:25:39 +02:00
committed by ahuston-0
parent 44b007c167
commit aee4e406e9
4 changed files with 87 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/hydra-evaluator/hydra-evaluator.cc
View File

@@ -180,10 +180,8 @@ struct Evaluator
{
auto conn(dbPool.get());
pqxx::work txn(*conn);
txn.exec_params0
("update Jobsets set startTime = $1 where id = $2",
now,
jobset.name.id);
txn.exec("update Jobsets set startTime = $1 where id = $2",
pqxx::params{now, jobset.name.id}).no_rows();
txn.commit();
}
@@ -234,7 +232,7 @@ struct Evaluator
pqxx::work txn(*conn);
if (jobset.evaluation_style == EvaluationStyle::ONE_AT_A_TIME) {
auto evaluation_res = txn.exec_params
auto evaluation_res = txn.exec
("select id from JobsetEvals "
"where jobset_id = $1 "
"order by id desc limit 1"
@@ -250,7 +248,7 @@ struct Evaluator
auto evaluation_id = evaluation_res[0][0].as<int>();
auto unfinished_build_res = txn.exec_params
auto unfinished_build_res = txn.exec
("select id from Builds "
"join JobsetEvalMembers "
" on (JobsetEvalMembers.build = Builds.id) "
@@ -420,21 +418,18 @@ struct Evaluator
/* Clear the trigger time to prevent this
jobset from getting stuck in an endless
failing eval loop. */
txn.exec_params0
txn.exec
("update Jobsets set triggerTime = null where id = $1 and startTime is not null and triggerTime <= startTime",
jobset.name.id);
jobset.name.id).no_rows();
/* Clear the start time. */
txn.exec_params0
txn.exec
("update Jobsets set startTime = null where id = $1",
jobset.name.id);
jobset.name.id).no_rows();
if (!WIFEXITED(status) || WEXITSTATUS(status) > 1) {
txn.exec_params0
("update Jobsets set errorMsg = $1, lastCheckedTime = $2, errorTime = $2, fetchErrorMsg = null where id = $3",
fmt("evaluation %s", statusToString(status)),
now,
jobset.name.id);
txn.exec("update Jobsets set errorMsg = $1, lastCheckedTime = $2, errorTime = $2, fetchErrorMsg = null where id = $3",
pqxx::params{fmt("evaluation %s", statusToString(status)), now, jobset.name.id}).no_rows();
}
txn.commit();
@@ -459,7 +454,7 @@ struct Evaluator
{
auto conn(dbPool.get());
pqxx::work txn(*conn);
txn.exec("update Jobsets set startTime = null");
txn.exec("update Jobsets set startTime = null").no_rows();
txn.commit();
}

9
src/hydra-queue-runner/builder.cc
View File

@@ -458,13 +458,12 @@ void State::failStep(
for (auto & build : indirect) {
if (build->finishedInDB) continue;
printError("marking build %1% as failed", build->id);
txn.exec_params0
("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $4, isCachedBuild = $5, notificationPendingSince = $4 where id = $1 and finished = 0",
build->id,
txn.exec("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $4, isCachedBuild = $5, notificationPendingSince = $4 where id = $1 and finished = 0",
pqxx::params{build->id,
(int) (build->drvPath != step->drvPath && result.buildStatus() == bsFailed ? bsDepFailed : result.buildStatus()),
result.startTime,
result.stopTime,
result.stepStatus == bsCachedFailure ? 1 : 0);
result.stepStatus == bsCachedFailure ? 1 : 0}).no_rows();
nrBuildsDone++;
}
@@ -473,7 +472,7 @@ void State::failStep(
if (result.stepStatus != bsCachedFailure && result.canCache)
for (auto & i : step->drv->outputsAndOptPaths(*localStore))
if (i.second.second)
txn.exec_params0("insert into FailedPaths values ($1)", localStore->printStorePath(*i.second.second));
txn.exec("insert into FailedPaths values ($1)", pqxx::params{localStore->printStorePath(*i.second.second)}).no_rows();
txn.commit();
}

109
src/hydra-queue-runner/hydra-queue-runner.cc
View File

@@ -276,17 +276,16 @@ void State::monitorMachinesFile()
void State::clearBusy(Connection & conn, time_t stopTime)
{
pqxx::work txn(conn);
txn.exec_params0
("update BuildSteps set busy = 0, status = $1, stopTime = $2 where busy != 0",
(int) bsAborted,
stopTime != 0 ? std::make_optional(stopTime) : std::nullopt);
txn.exec("update BuildSteps set busy = 0, status = $1, stopTime = $2 where busy != 0",
pqxx::params{(int) bsAborted,
stopTime != 0 ? std::make_optional(stopTime) : std::nullopt}).no_rows();
txn.commit();
}
unsigned int State::allocBuildStep(pqxx::work & txn, BuildID buildId)
{
auto res = txn.exec_params1("select max(stepnr) from BuildSteps where build = $1", buildId);
auto res = txn.exec("select max(stepnr) from BuildSteps where build = $1", buildId).one_row();
return res[0].is_null() ? 1 : res[0].as<int>() + 1;
}
@@ -297,9 +296,8 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
restart:
auto stepNr = allocBuildStep(txn, buildId);
auto r = txn.exec_params
("insert into BuildSteps (build, stepnr, type, drvPath, busy, startTime, system, status, propagatedFrom, errorMsg, stopTime, machine) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) on conflict do nothing",
buildId,
auto r = txn.exec("insert into BuildSteps (build, stepnr, type, drvPath, busy, startTime, system, status, propagatedFrom, errorMsg, stopTime, machine) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) on conflict do nothing",
pqxx::params{buildId,
stepNr,
0, // == build
localStore->printStorePath(step->drvPath),
@@ -310,17 +308,16 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
propagatedFrom != 0 ? std::make_optional(propagatedFrom) : std::nullopt, // internal::params
errorMsg != "" ? std::make_optional(errorMsg) : std::nullopt,
startTime != 0 && status != bsBusy ? std::make_optional(startTime) : std::nullopt,
machine);
machine});
if (r.affected_rows() == 0) goto restart;
for (auto & [name, output] : getDestStore()->queryPartialDerivationOutputMap(step->drvPath, &*localStore))
txn.exec_params0
("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
buildId, stepNr, name,
txn.exec("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
pqxx::params{buildId, stepNr, name,
output
? std::optional { localStore->printStorePath(*output)}
: std::nullopt);
: std::nullopt}).no_rows();
if (status == bsBusy)
txn.exec(fmt("notify step_started, '%d\t%d'", buildId, stepNr));
@@ -331,11 +328,10 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
void State::updateBuildStep(pqxx::work & txn, BuildID buildId, unsigned int stepNr, StepState stepState)
{
if (txn.exec_params
("update BuildSteps set busy = $1 where build = $2 and stepnr = $3 and busy != 0 and status is null",
(int) stepState,
if (txn.exec("update BuildSteps set busy = $1 where build = $2 and stepnr = $3 and busy != 0 and status is null",
pqxx::params{(int) stepState,
buildId,
stepNr).affected_rows() != 1)
stepNr}).affected_rows() != 1)
throw Error("step %d of build %d is in an unexpected state", stepNr, buildId);
}
@@ -345,29 +341,27 @@ void State::finishBuildStep(pqxx::work & txn, const RemoteResult & result,
{
assert(result.startTime);
assert(result.stopTime);
txn.exec_params0
("update BuildSteps set busy = 0, status = $1, errorMsg = $4, startTime = $5, stopTime = $6, machine = $7, overhead = $8, timesBuilt = $9, isNonDeterministic = $10 where build = $2 and stepnr = $3",
(int) result.stepStatus, buildId, stepNr,
txn.exec("update BuildSteps set busy = 0, status = $1, errorMsg = $4, startTime = $5, stopTime = $6, machine = $7, overhead = $8, timesBuilt = $9, isNonDeterministic = $10 where build = $2 and stepnr = $3",
pqxx::params{(int) result.stepStatus, buildId, stepNr,
result.errorMsg != "" ? std::make_optional(result.errorMsg) : std::nullopt,
result.startTime, result.stopTime,
machine != "" ? std::make_optional(machine) : std::nullopt,
result.overhead != 0 ? std::make_optional(result.overhead) : std::nullopt,
result.timesBuilt > 0 ? std::make_optional(result.timesBuilt) : std::nullopt,
result.timesBuilt > 1 ? std::make_optional(result.isNonDeterministic) : std::nullopt);
result.timesBuilt > 1 ? std::make_optional(result.isNonDeterministic) : std::nullopt}).no_rows();
assert(result.logFile.find('\t') == std::string::npos);
txn.exec(fmt("notify step_finished, '%d\t%d\t%s'",
buildId, stepNr, result.logFile));
if (result.stepStatus == bsSuccess) {
// Update the corresponding `BuildStepOutputs` row to add the output path
auto res = txn.exec_params1("select drvPath from BuildSteps where build = $1 and stepnr = $2", buildId, stepNr);
auto res = txn.exec("select drvPath from BuildSteps where build = $1 and stepnr = $2", pqxx::params{buildId, stepNr}).one_row();
assert(res.size());
StorePath drvPath = localStore->parseStorePath(res[0].as<std::string>());
// If we've finished building, all the paths should be known
for (auto & [name, output] : getDestStore()->queryDerivationOutputMap(drvPath, &*localStore))
txn.exec_params0
("update BuildStepOutputs set path = $4 where build = $1 and stepnr = $2 and name = $3",
buildId, stepNr, name, localStore->printStorePath(output));
txn.exec("update BuildStepOutputs set path = $4 where build = $1 and stepnr = $2 and name = $3",
pqxx::params{buildId, stepNr, name, localStore->printStorePath(output)}).no_rows();
}
}
@@ -378,23 +372,21 @@ int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t sto
restart:
auto stepNr = allocBuildStep(txn, build->id);
auto r = txn.exec_params
("insert into BuildSteps (build, stepnr, type, drvPath, busy, status, startTime, stopTime) values ($1, $2, $3, $4, $5, $6, $7, $8) on conflict do nothing",
build->id,
auto r = txn.exec("insert into BuildSteps (build, stepnr, type, drvPath, busy, status, startTime, stopTime) values ($1, $2, $3, $4, $5, $6, $7, $8) on conflict do nothing",
pqxx::params{build->id,
stepNr,
1, // == substitution
(localStore->printStorePath(drvPath)),
0,
0,
startTime,
stopTime);
stopTime});
if (r.affected_rows() == 0) goto restart;
txn.exec_params0
("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
build->id, stepNr, outputName,
localStore->printStorePath(storePath));
txn.exec("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
pqxx::params{build->id, stepNr, outputName,
localStore->printStorePath(storePath)}).no_rows();
return stepNr;
}
@@ -461,35 +453,32 @@ void State::markSucceededBuild(pqxx::work & txn, Build::ptr build,
{
if (build->finishedInDB) return;
if (txn.exec_params("select 1 from Builds where id = $1 and finished = 0", build->id).empty()) return;
if (txn.exec("select 1 from Builds where id = $1 and finished = 0", pqxx::params{build->id}).empty()) return;
txn.exec_params0
("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $4, size = $5, closureSize = $6, releaseName = $7, isCachedBuild = $8, notificationPendingSince = $4 where id = $1",
build->id,
txn.exec("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $4, size = $5, closureSize = $6, releaseName = $7, isCachedBuild = $8, notificationPendingSince = $4 where id = $1",
pqxx::params{build->id,
(int) (res.failed ? bsFailedWithOutput : bsSuccess),
startTime,
stopTime,
res.size,
res.closureSize,
res.releaseName != "" ? std::make_optional(res.releaseName) : std::nullopt,
isCachedBuild ? 1 : 0);
isCachedBuild ? 1 : 0}).no_rows();
for (auto & [outputName, outputPath] : res.outputs) {
txn.exec_params0
("update BuildOutputs set path = $3 where build = $1 and name = $2",
build->id,
txn.exec("update BuildOutputs set path = $3 where build = $1 and name = $2",
pqxx::params{build->id,
outputName,
localStore->printStorePath(outputPath)
);
localStore->printStorePath(outputPath)}
).no_rows();
}
txn.exec_params0("delete from BuildProducts where build = $1", build->id);
txn.exec("delete from BuildProducts where build = $1", pqxx::params{build->id}).no_rows();
unsigned int productNr = 1;
for (auto & product : res.products) {
txn.exec_params0
("insert into BuildProducts (build, productnr, type, subtype, fileSize, sha256hash, path, name, defaultPath) values ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
build->id,
txn.exec("insert into BuildProducts (build, productnr, type, subtype, fileSize, sha256hash, path, name, defaultPath) values ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
pqxx::params{build->id,
productNr++,
product.type,
product.subtype,
@@ -497,22 +486,21 @@ void State::markSucceededBuild(pqxx::work & txn, Build::ptr build,
product.sha256hash ? std::make_optional(product.sha256hash->to_string(HashFormat::Base16, false)) : std::nullopt,
product.path,
product.name,
product.defaultPath);
product.defaultPath}).no_rows();
}
txn.exec_params0("delete from BuildMetrics where build = $1", build->id);
txn.exec("delete from BuildMetrics where build = $1", pqxx::params{build->id}).no_rows();
for (auto & metric : res.metrics) {
txn.exec_params0
("insert into BuildMetrics (build, name, unit, value, project, jobset, job, timestamp) values ($1, $2, $3, $4, $5, $6, $7, $8)",
build->id,
txn.exec("insert into BuildMetrics (build, name, unit, value, project, jobset, job, timestamp) values ($1, $2, $3, $4, $5, $6, $7, $8)",
pqxx::params{build->id,
metric.second.name,
metric.second.unit != "" ? std::make_optional(metric.second.unit) : std::nullopt,
metric.second.value,
build->projectName,
build->jobsetName,
build->jobName,
build->timestamp);
build->timestamp}).no_rows();
}
nrBuildsDone++;
@@ -524,7 +512,7 @@ bool State::checkCachedFailure(Step::ptr step, Connection & conn)
pqxx::work txn(conn);
for (auto & i : step->drv->outputsAndOptPaths(*localStore))
if (i.second.second)
if (!txn.exec_params("select 1 from FailedPaths where path = $1", localStore->printStorePath(*i.second.second)).empty())
if (!txn.exec("select 1 from FailedPaths where path = $1", pqxx::params{localStore->printStorePath(*i.second.second)}).empty())
return true;
return false;
}
@@ -736,8 +724,8 @@ void State::dumpStatus(Connection & conn)
auto mc = startDbUpdate();
pqxx::work txn(conn);
// FIXME: use PostgreSQL 9.5 upsert.
txn.exec("delete from SystemStatus where what = 'queue-runner'");
txn.exec_params0("insert into SystemStatus values ('queue-runner', $1)", statusJson.dump());
txn.exec("delete from SystemStatus where what = 'queue-runner'").no_rows();
txn.exec("insert into SystemStatus values ('queue-runner', $1)", pqxx::params{statusJson.dump()}).no_rows();
txn.exec("notify status_dumped");
txn.commit();
}
@@ -802,7 +790,7 @@ void State::unlock()
{
pqxx::work txn(*conn);
txn.exec("delete from SystemStatus where what = 'queue-runner'");
txn.exec("delete from SystemStatus where what = 'queue-runner'").no_rows();
txn.commit();
}
}
@@ -880,11 +868,10 @@ void State::run(BuildID buildOne)
pqxx::work txn(*conn);
for (auto & step : steps) {
printMsg(lvlError, "cleaning orphaned step %d of build %d", step.second, step.first);
txn.exec_params0
("update BuildSteps set busy = 0, status = $1 where build = $2 and stepnr = $3 and busy != 0",
(int) bsAborted,
txn.exec("update BuildSteps set busy = 0, status = $1 where build = $2 and stepnr = $3 and busy != 0",
pqxx::params{(int) bsAborted,
step.first,
step.second);
step.second}).no_rows();
}
txn.commit();
} catch (std::exception & e) {

59
src/hydra-queue-runner/queue-monitor.cc
View File

@@ -108,8 +108,7 @@ bool State::getQueuedBuilds(Connection & conn,
{
pqxx::work txn(conn);
auto res = txn.exec_params
("select builds.id, builds.jobset_id, jobsets.project as project, "
auto res = txn.exec("select builds.id, builds.jobset_id, jobsets.project as project, "
"jobsets.name as jobset, job, drvPath, maxsilent, timeout, timestamp, "
"globalPriority, priority from Builds "
"inner join jobsets on builds.jobset_id = jobsets.id "
@@ -158,11 +157,10 @@ bool State::getQueuedBuilds(Connection & conn,
if (!build->finishedInDB) {
auto mc = startDbUpdate();
pqxx::work txn(conn);
txn.exec_params0
("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $3 where id = $1 and finished = 0",
build->id,
txn.exec("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $3 where id = $1 and finished = 0",
pqxx::params{build->id,
(int) bsAborted,
time(0));
time(0)}).no_rows();
txn.commit();
build->finishedInDB = true;
nrBuildsDone++;
@@ -192,22 +190,20 @@ bool State::getQueuedBuilds(Connection & conn,
derivation path, then by output path. */
BuildID propagatedFrom = 0;
auto res = txn.exec_params1
("select max(build) from BuildSteps where drvPath = $1 and startTime != 0 and stopTime != 0 and status = 1",
localStore->printStorePath(ex.step->drvPath));
auto res = txn.exec("select max(build) from BuildSteps where drvPath = $1 and startTime != 0 and stopTime != 0 and status = 1",
pqxx::params{localStore->printStorePath(ex.step->drvPath)}).one_row();
if (!res[0].is_null()) propagatedFrom = res[0].as<BuildID>();
if (!propagatedFrom) {
for (auto & [outputName, optOutputPath] : destStore->queryPartialDerivationOutputMap(ex.step->drvPath, &*localStore)) {
constexpr std::string_view common = "select max(s.build) from BuildSteps s join BuildStepOutputs o on s.build = o.build where startTime != 0 and stopTime != 0 and status = 1";
auto res = optOutputPath
? txn.exec_params(
? txn.exec(
std::string { common } + " and path = $1",
localStore->printStorePath(*optOutputPath))
: txn.exec_params(
pqxx::params{localStore->printStorePath(*optOutputPath)})
: txn.exec(
std::string { common } + " and drvPath = $1 and name = $2",
localStore->printStorePath(ex.step->drvPath),
outputName);
pqxx::params{localStore->printStorePath(ex.step->drvPath), outputName});
if (!res[0][0].is_null()) {
propagatedFrom = res[0][0].as<BuildID>();
break;
@@ -216,12 +212,11 @@ bool State::getQueuedBuilds(Connection & conn,
}
createBuildStep(txn, 0, build->id, ex.step, "", bsCachedFailure, "", propagatedFrom);
txn.exec_params
("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $3, isCachedBuild = 1, notificationPendingSince = $3 "
txn.exec("update Builds set finished = 1, buildStatus = $2, startTime = $3, stopTime = $3, isCachedBuild = 1, notificationPendingSince = $3 "
"where id = $1 and finished = 0",
build->id,
pqxx::params{build->id,
(int) (ex.step->drvPath == build->drvPath ? bsFailed : bsDepFailed),
time(0));
time(0)}).no_rows();
notifyBuildFinished(txn, build->id, {});
txn.commit();
build->finishedInDB = true;
@@ -653,10 +648,8 @@ Jobset::ptr State::createJobset(pqxx::work & txn,
if (i != jobsets_->end()) return i->second;
}
auto res = txn.exec_params1
("select schedulingShares from Jobsets where id = $1",
jobsetID);
if (res.empty()) throw Error("missing jobset - can't happen");
auto res = txn.exec("select schedulingShares from Jobsets where id = $1",
pqxx::params{jobsetID}).one_row();
auto shares = res["schedulingShares"].as<unsigned int>();
@@ -664,11 +657,10 @@ Jobset::ptr State::createJobset(pqxx::work & txn,
jobset->setShares(shares);
/* Load the build steps from the last 24 hours. */
auto res2 = txn.exec_params
("select s.startTime, s.stopTime from BuildSteps s join Builds b on build = id "
auto res2 = txn.exec("select s.startTime, s.stopTime from BuildSteps s join Builds b on build = id "
"where s.startTime is not null and s.stopTime > $1 and jobset_id = $2",
time(0) - Jobset::schedulingWindow * 10,
jobsetID);
pqxx::params{time(0) - Jobset::schedulingWindow * 10,
jobsetID});
for (auto const & row : res2) {
time_t startTime = row["startTime"].as<time_t>();
time_t stopTime = row["stopTime"].as<time_t>();
@@ -705,11 +697,10 @@ BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store>
pqxx::work txn(conn);
for (auto & [name, output] : derivationOutputs) {
auto r = txn.exec_params
("select id, buildStatus, releaseName, closureSize, size from Builds b "
auto r = txn.exec("select id, buildStatus, releaseName, closureSize, size from Builds b "
"join BuildOutputs o on b.id = o.build "
"where finished = 1 and (buildStatus = 0 or buildStatus = 6) and path = $1",
localStore->printStorePath(output));
pqxx::params{localStore->printStorePath(output)});
if (r.empty()) continue;
BuildID id = r[0][0].as<BuildID>();
@@ -721,9 +712,8 @@ BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store>
res.closureSize = r[0][3].is_null() ? 0 : r[0][3].as<uint64_t>();
res.size = r[0][4].is_null() ? 0 : r[0][4].as<uint64_t>();
auto products = txn.exec_params
("select type, subtype, fileSize, sha256hash, path, name, defaultPath from BuildProducts where build = $1 order by productnr",
id);
auto products = txn.exec("select type, subtype, fileSize, sha256hash, path, name, defaultPath from BuildProducts where build = $1 order by productnr",
pqxx::params{id});
for (auto row : products) {
BuildProduct product;
@@ -745,9 +735,8 @@ BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store>
res.products.emplace_back(product);
}
auto metrics = txn.exec_params
("select name, unit, value from BuildMetrics where build = $1",
id);
auto metrics = txn.exec("select name, unit, value from BuildMetrics where build = $1",
pqxx::params{id});
for (auto row : metrics) {
BuildMetric metric;