talexander/nix_builder
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Merge verify nix store before builds.

Browse Source
This commit is contained in:
Tom Alexander
2026-09-03 22:13:39 -04:00
parent 3aad35950e 0a270dcbdd
commit c994952ab3
6 changed files with 298 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
example_config.toml
View File

@@ -1,5 +1,12 @@
# repo_directory = "/home/nixworker/persist/nix_builder"
[[targets]]
name = "foo"
repo = "https://code.fizz.buzz/talexander/machine_setup.git"
branch = "foo"
path = "nix/configuration"
attr = "nixosConfigurations.odo.config.system.build.toplevel"
[[targets]]
name = "odo"
repo = "https://code.fizz.buzz/talexander/machine_setup.git"

88
src/command/build/runner.rs
View File

@@ -1,4 +1,7 @@
use std::process::ExitStatus;
use sqlx::Row;
use tracing::error;
use tracing::info;
use crate::Result;
@@ -12,6 +15,7 @@ use crate::git_util::git_force_into_state;
use crate::git_util::git_init_at_rev;
use crate::nix_util::FlakeLock;
use crate::nix_util::nix_flake_update;
use crate::nix_util::nix_store_verify_repair_check_contents;
use crate::nix_util::nixos_build_target;
use crate::nix_util::parse_flake_lock;
@@ -33,7 +37,62 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
let db_handle = DbHandle::new(Some(database_path)).await?;
verify_nix_store().await?;
for target_name in args.target {
// Record start of build
let build_id: i64 = sqlx::query(
r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#,
)
.bind(&target_name)
.fetch_one(&db_handle.conn)
.await?
.try_get("id")?;
// Put the rest into a function
let result = full_build_target(&config, &db_handle, &target_name, build_id).await;
match result {
Ok(exit_status) => {
let update: u64 = sqlx::query(
r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#,
)
.bind(
exit_status
.code()
.expect("Process should have an exit code."),
)
.bind(build_id)
.execute(&db_handle.conn)
.await?
.rows_affected();
assert!(update == 1);
}
Err(e) => {
error!("Error building target {}: {}", target_name, e);
let update: u64 = sqlx::query(
r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#,
)
.bind(-1)
.bind(build_id)
.execute(&db_handle.conn)
.await?
.rows_affected();
assert!(update == 1);
}
};
}
db_handle.conn.close().await;
Ok(())
}
async fn full_build_target(
config: &Config,
db_handle: &DbHandle,
target_name: &str,
build_id: i64,
) -> Result<ExitStatus> {
let target_config = {
let target_config = config.get_target_config(&target_name)?;
if let Some(conf) = target_config {
@@ -47,12 +106,8 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
if target_config.get_update() {
run_nix_update(&config, target_config).await?;
}
build_target(&db_handle, &config, target_config).await?;
}
db_handle.conn.close().await;
Ok(())
let exit_status = build_target(&db_handle, &config, target_config, build_id).await?;
Ok(exit_status)
}
async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig) -> Result<()> {
@@ -85,6 +140,12 @@ async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig)
Ok(())
}
async fn verify_nix_store() -> Result<()> {
nix_store_verify_repair_check_contents().await?;
Ok(())
}
async fn run_nix_update(config_root: &Config, target_config: &TargetConfig) -> Result<()> {
let flake_directory = target_config.get_flake_directory(config_root)?;
@@ -97,7 +158,8 @@ async fn build_target(
db_handle: &DbHandle,
config_root: &Config,
target_config: &TargetConfig,
) -> Result<()> {
build_id: i64,
) -> Result<ExitStatus> {
let flake_directory = target_config.get_flake_directory(config_root)?;
let build_directory = target_config.get_build_directory(config_root)?;
assert_directory!(
@@ -108,18 +170,10 @@ async fn build_target(
let target_name = target_config.get_name()?;
let build_id: i64 = sqlx::query(
r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#,
)
.bind(target_name)
.fetch_one(&db_handle.conn)
.await?
.try_get("id")?;
let flake_lock = parse_flake_lock(&flake_directory).await?;
write_input_revs_to_db(db_handle, &flake_lock, build_id).await?;
nixos_build_target(
let exit_status = nixos_build_target(
db_handle,
build_directory,
flake_directory,
@@ -129,7 +183,7 @@ async fn build_target(
)
.await?;
Ok(())
Ok(exit_status)
}
async fn write_input_revs_to_db(

31
src/nix_util/high_level.rs
View File

@@ -1,6 +1,7 @@
use std::ffi::OsStr;
use std::ffi::OsString;
use std::path::Path;
use std::process::ExitStatus;
use std::process::Stdio;
use tokio::process::Command;
@@ -9,6 +10,7 @@ use crate::Result;
use crate::database::db_handle::DbHandle;
use super::RunningUpdate;
use super::RunningVerify;
use super::running_build::RunningBuild;
pub(crate) async fn nixos_build_target<B, F, A, TN>(
@@ -18,7 +20,7 @@ pub(crate) async fn nixos_build_target<B, F, A, TN>(
attr: A,
target_name: TN,
build_id: i64,
) -> Result<()>
) -> Result<ExitStatus>
where
B: AsRef<Path>,
F: AsRef<Path>,
@@ -57,9 +59,9 @@ where
let child = command.spawn()?;
let mut running_build = RunningBuild::new(db_handle, build_id)?;
running_build.run_to_completion(child, target_name).await?;
let exit_status = running_build.run_to_completion(child, target_name).await?;
Ok(())
Ok(exit_status)
}
pub(crate) async fn nix_flake_update<F>(flake_path: F) -> Result<()>
@@ -87,3 +89,26 @@ where
Ok(())
}
pub(crate) async fn nix_store_verify_repair_check_contents() -> Result<()> {
let mut command = Command::new("nix-store");
command.stdout(Stdio::piped());
command.stderr(Stdio::piped());
command.stdin(Stdio::null());
command.args([
"--verify",
"--check-contents",
"--repair",
"--log-format",
"internal-json",
"-vvvvvvvvvvv",
]);
command.kill_on_drop(true);
let child = command.spawn()?;
let mut running_verify = RunningVerify::new()?;
running_verify.run_to_completion(child).await?;
Ok(())
}

2
src/nix_util/mod.rs
View File

@@ -7,6 +7,7 @@ mod nix_output_stream;
mod output_stream;
mod running_build;
mod running_update;
mod running_verify;
mod tree_iter;
pub(crate) use activity_tree::ActivityIdAlreadyInTreeError;
pub(crate) use activity_tree::ActivityIdNotInTreeError;
@@ -18,3 +19,4 @@ pub(crate) use output_stream::OutputLine;
pub(crate) use output_stream::OutputLineStream;
pub(crate) use running_build::RunningBuild;
pub(crate) use running_update::RunningUpdate;
pub(crate) use running_verify::RunningVerify;

18
src/nix_util/running_build.rs
View File

@@ -1,4 +1,5 @@
use std::borrow::Cow;
use std::process::ExitStatus;
use std::time::Duration;
use std::time::Instant;
use std::time::SystemTime;
@@ -43,7 +44,7 @@ impl<'db> RunningBuild<'db> {
&mut self,
mut child: Child,
target_name: TN,
) -> Result<()>
) -> Result<ExitStatus>
where
TN: AsRef<str>,
{
@@ -66,20 +67,7 @@ impl<'db> RunningBuild<'db> {
let exit_status = exit_status_handle.await?;
println!("nix build status was: {}", exit_status);
let update: u64 =
sqlx::query(r#"UPDATE build SET end_time=unixepoch('now'), status=? WHERE id=?"#)
.bind(
exit_status
.code()
.expect("Process should have an exit code."),
)
.bind(self.build_id)
.execute(&self.db_handle.conn)
.await?
.rows_affected();
assert!(update == 1);
Ok(())
Ok(exit_status)
}
pub(crate) fn handle_message(&mut self, message: NixMessage) -> Result<()> {

181
src/nix_util/running_verify.rs Normal file
View File

@@ -0,0 +1,181 @@
use std::borrow::Cow;
use std::time::Duration;
use std::time::Instant;
use tokio::process::Child;
use tracing::error;
use tracing::info;
use crate::Result;
use crate::nix_util::NixOutputStream;
use crate::nix_util::nix_output_stream::NixAction;
use crate::nix_util::output_stream::OutputStream;
use crate::nix_util::tree_iter::ForwardTreeIter;
use super::activity_tree::ActivityTreeEntry;
use super::activity_tree_stream::ActivityTreeStream;
use super::nix_output_stream::ActivityResultMessage;
use super::nix_output_stream::NixMessage;
pub(crate) struct RunningVerify {
activity_tree: ActivityTreeStream,
last_announce: Option<Instant>,
}
impl RunningVerify {
pub(crate) fn new() -> Result<Self> {
Ok(RunningVerify {
activity_tree: ActivityTreeStream::new(),
last_announce: None,
})
}
pub(crate) async fn run_to_completion(&mut self, mut child: Child) -> Result<()> {
let output_stream = OutputStream::from_child(&mut child)?;
let mut nix_output_stream: NixOutputStream<OutputStream> =
NixOutputStream::new(output_stream);
info!("Verifying nix store.");
let exit_status_handle = tokio::spawn(async move {
let status = child
.wait()
.await
.expect("nixos-rebuild encountered an error");
status
});
while let Some(message) = nix_output_stream.next().await? {
self.handle_message(message)?;
}
let exit_status = exit_status_handle.await?;
info!("nix store verify status was: {}", exit_status);
Ok(())
}
pub(crate) fn handle_message(&mut self, message: NixMessage) -> Result<()> {
self.activity_tree.handle_message(&message)?;
let message = match message {
NixMessage::ParseFailure(line) => {
error!("FAIL PARSE: {line}");
return Ok(());
}
NixMessage::Generic(_value, line) => {
error!("GENERIC PARSE: {line}");
return Ok(());
}
NixMessage::Action(nix_action) => nix_action,
};
match message {
NixAction::Msg(msg_message) => {
if msg_message.level > 0 && msg_message.level < 4 {
eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg);
}
}
NixAction::Start(activity_start_message) => {
// println!("START: {}", serde_json::to_string(&activity_start_message)?);
self.print_current_status();
}
NixAction::Stop(stop_message) => {
// println!("STOP: {}", serde_json::to_string(&stop_message)?);
self.print_current_status();
}
NixAction::Result(activity_result_message) => {
match activity_result_message {
ActivityResultMessage::FileLinked(_activity_result_file_linked) => {}
ActivityResultMessage::BuildLogLine(_activity_result_build_log_line) => {}
ActivityResultMessage::UntrustedPath(_activity_result_untrusted_path) => {}
ActivityResultMessage::CorruptedPath(_activity_result_corrupted_path) => {}
ActivityResultMessage::SetPhase(_activity_result_set_phase) => {}
ActivityResultMessage::Progress(activity_result_progress) => {
// if activity_result_progress.expected != 0 {
// println!(
// "PROGRESS: {}",
// serde_json::to_string(&activity_result_progress)?
// );
// }
self.maybe_print_current_status();
}
ActivityResultMessage::SetExpected(activity_result_set_expected) => {
// if activity_result_set_expected.expected != 0 {
// println!(
// "EXPECTED: {}",
// serde_json::to_string(&activity_result_set_expected)?
// );
// }
self.maybe_print_current_status();
}
ActivityResultMessage::PostBuildLogLine(
_activity_result_post_build_log_line,
) => {}
ActivityResultMessage::FetchStatus(_activity_result_fetch_status) => {}
};
}
};
Ok(())
}
fn maybe_print_current_status(&mut self) -> () {
let last_announce = match self.last_announce {
Some(instant) => instant,
None => {
// If we haven't announced before, always announce.
return self.print_current_status();
}
};
let now = Instant::now();
let time_since_last_announce = now.duration_since(last_announce);
if time_since_last_announce > Duration::new(5, 0) {
return self.print_current_status();
}
}
fn print_current_status(&mut self) -> () {
let nodes = ForwardTreeIter::new(
self.activity_tree.get_tree(),
None,
is_match_predicate,
is_transparent_predicate,
is_alive_predicate,
);
let mut out = Vec::new();
for (depth, _is_match, _is_transparent, node) in nodes {
let progress_text = node
.get_activity()
.get_progress_text()
.unwrap_or(Cow::Borrowed(""));
let name = node
.get_activity()
.display_name()
.unwrap_or(Cow::Borrowed("null"));
out.push(format!("{depth}\t{progress_text}\t{name}"))
}
if out.is_empty() {
println!("No active activities.");
} else {
println!("\n");
for l in out {
println!("{l}\n");
}
println!("\n");
}
self.last_announce = Some(Instant::now());
}
}
fn is_match_predicate(entry: &ActivityTreeEntry) -> bool {
entry.get_activity().get_progress_text().is_some()
}
pub(crate) fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool {
return false;
}
fn is_alive_predicate(entry: &ActivityTreeEntry) -> bool {
entry.get_activity().is_active()
}