Compare commits

...

9 Commits

Author SHA1 Message Date
Tom Alexander
6386febdb8 Verify the nix store before starting a build run. 2026-08-04 21:36:38 -04:00
Tom Alexander
3aad35950e Clean up RunningUpdate. 2026-08-04 21:36:37 -04:00
Tom Alexander
83f05b3677 Add support for tar in flake.lock. 2026-08-04 21:35:30 -04:00
Tom Alexander
0addf91edf Add support for parsing the flake.lock. 2026-07-28 10:19:23 -04:00
Tom Alexander
59fdbf41ef Do not print the "checking path" messages. 2026-07-18 14:06:51 -04:00
Tom Alexander
f59e45998e Do not print the "evaluating file" messages. 2026-07-18 14:03:23 -04:00
Tom Alexander
3bf962de00 Do not print the "linking" messages. 2026-07-18 11:29:35 -04:00
Tom Alexander
64471269f2 Always render a tree when an activity reaches 100%. 2026-07-18 11:23:11 -04:00
Tom Alexander
0b2de9d40c Support progress for VerifyPaths. 2026-07-18 11:09:48 -04:00
13 changed files with 531 additions and 45 deletions

2
.gitignore vendored
View File

@@ -3,3 +3,5 @@
/example_logs /example_logs
TODO.org TODO.org
/result /result
/.envrc
/.direnv/

View File

@@ -0,0 +1,2 @@
DROP INDEX ix_flake_inputs_build_id;
DROP TABLE flake_inputs;

View File

@@ -0,0 +1,9 @@
CREATE TABLE flake_inputs (
id INTEGER NOT NULL PRIMARY KEY,
build_id INTEGER NOT NULL,
input_name TEXT NOT NULL,
rev TEXT NOT NULL,
FOREIGN KEY(build_id) REFERENCES build(id)
) STRICT;
CREATE INDEX ix_flake_inputs_build_id ON flake_inputs(build_id);

View File

@@ -1,3 +1,6 @@
use sqlx::Row;
use tracing::info;
use crate::Result; use crate::Result;
use crate::cli::parameters::BuildArgs; use crate::cli::parameters::BuildArgs;
use crate::config::Config; use crate::config::Config;
@@ -7,8 +10,11 @@ use crate::fs_util::assert_directory;
use crate::fs_util::is_git_repo; use crate::fs_util::is_git_repo;
use crate::git_util::git_force_into_state; use crate::git_util::git_force_into_state;
use crate::git_util::git_init_at_rev; 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_flake_update;
use crate::nix_util::nix_store_verify_repair_check_contents;
use crate::nix_util::nixos_build_target; use crate::nix_util::nixos_build_target;
use crate::nix_util::parse_flake_lock;
pub(crate) async fn run_build(args: BuildArgs) -> Result<()> { pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
println!("{:?}", args); println!("{:?}", args);
@@ -28,6 +34,8 @@ pub(crate) async fn run_build(args: BuildArgs) -> Result<()> {
let db_handle = DbHandle::new(Some(database_path)).await?; let db_handle = DbHandle::new(Some(database_path)).await?;
verify_nix_store().await?;
for target_name in args.target { for target_name in args.target {
let target_config = { let target_config = {
let target_config = config.get_target_config(&target_name)?; let target_config = config.get_target_config(&target_name)?;
@@ -80,6 +88,12 @@ async fn prepare_flake_repo(config_root: &Config, target_config: &TargetConfig)
Ok(()) 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<()> { async fn run_nix_update(config_root: &Config, target_config: &TargetConfig) -> Result<()> {
let flake_directory = target_config.get_flake_directory(config_root)?; let flake_directory = target_config.get_flake_directory(config_root)?;
@@ -101,14 +115,61 @@ async fn build_target(
build_directory.to_string_lossy() build_directory.to_string_lossy()
); );
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( nixos_build_target(
db_handle, db_handle,
build_directory, build_directory,
flake_directory, flake_directory,
target_config.get_attr()?, target_config.get_attr()?,
target_config.get_name()?, target_name,
build_id,
) )
.await?; .await?;
Ok(()) Ok(())
} }
async fn write_input_revs_to_db(
db_handle: &DbHandle,
flake_lock: &FlakeLock,
build_id: i64,
) -> Result<()> {
let root_node = flake_lock.get_root_node()?;
let direct_inputs = flake_lock.get_inputs(root_node)?;
let mut tx = db_handle.conn.begin().await?;
for (input, node) in direct_inputs.iter() {
let locked = match &node.locked {
Some(locked) => locked,
None => {
info!(
"Not recording rev for input {} because it lacks a locked section.",
input
);
continue;
}
};
let rev = &locked.rev;
info!("input {} {}", input, rev);
sqlx::query(r#"INSERT INTO flake_inputs (build_id, input_name, rev) VALUES (?, ?, ?);"#)
.bind(build_id)
.bind(input)
.bind(rev)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}

View File

@@ -14,12 +14,10 @@ use crate::nix_util::NixOutputStream;
use crate::nix_util::OutputLine; use crate::nix_util::OutputLine;
use crate::nix_util::OutputLineStream; use crate::nix_util::OutputLineStream;
use crate::nix_util::RunningBuild; use crate::nix_util::RunningBuild;
use crate::nix_util::RunningUpdate;
pub(crate) async fn feed_logs(args: FeedLogArgs) -> Result<()> { pub(crate) async fn feed_logs(args: FeedLogArgs) -> Result<()> {
let db_handle = DbHandle::new::<String>(None).await?; let db_handle = DbHandle::new::<String>(None).await?;
// let mut running_build = RunningUpdate::new()?; let mut running_build = RunningBuild::new(&db_handle, -1)?;
let mut running_build = RunningBuild::new(&db_handle)?;
let file_stream = FileStream::new(args.input).await?; let file_stream = FileStream::new(args.input).await?;
let mut nix_output_stream = NixOutputStream::new(file_stream); let mut nix_output_stream = NixOutputStream::new(file_stream);
while let Some(message) = nix_output_stream.next().await? { while let Some(message) = nix_output_stream.next().await? {

View File

@@ -215,7 +215,15 @@ impl Activity {
.as_ref() .as_ref()
.map(|phase| Cow::Owned(format!("[{}]", phase))), .map(|phase| Cow::Owned(format!("[{}]", phase))),
Activity::OptimizeStore(_activity_optimize_store) => None, Activity::OptimizeStore(_activity_optimize_store) => None,
Activity::VerifyPaths(_activity_verify_paths) => None, Activity::VerifyPaths(activity_verify_paths) => {
get_progress_bar(activity_verify_paths.done, activity_verify_paths.expected)
.or_else(|| {
get_progress_text(
activity_verify_paths.done,
activity_verify_paths.expected,
)
})
}
Activity::Substitute(_activity_substitute) => None, Activity::Substitute(_activity_substitute) => None,
Activity::QueryPathInfo(_activity_query_path_info) => None, Activity::QueryPathInfo(_activity_query_path_info) => None,
Activity::PostBuildHook(_activity_post_build_hook) => None, Activity::PostBuildHook(_activity_post_build_hook) => None,
@@ -323,8 +331,11 @@ impl Activity {
Activity::OptimizeStore(_activity_optimize_store) => { Activity::OptimizeStore(_activity_optimize_store) => {
panic!("Attempted to set the progress of an optimize store activity."); panic!("Attempted to set the progress of an optimize store activity.");
} }
Activity::VerifyPaths(_activity_verify_paths) => { Activity::VerifyPaths(activity_verify_paths) => {
panic!("Attempted to set the progress of a verify paths activity."); activity_verify_paths.done = done;
activity_verify_paths.expected = expected;
activity_verify_paths.running = running;
activity_verify_paths.failed = failed;
} }
Activity::Substitute(_activity_substitute) => { Activity::Substitute(_activity_substitute) => {
panic!("Attempted to set the progress of a substitute activity."); panic!("Attempted to set the progress of a substitute activity.");
@@ -451,6 +462,10 @@ pub(crate) struct ActivityOptimizeStore {
} }
pub(crate) struct ActivityVerifyPaths { pub(crate) struct ActivityVerifyPaths {
pub(crate) state: ActivityState, pub(crate) state: ActivityState,
pub(crate) done: u64,
pub(crate) expected: u64,
pub(crate) running: u64,
pub(crate) failed: u64,
} }
pub(crate) struct ActivitySubstitute { pub(crate) struct ActivitySubstitute {
pub(crate) state: ActivityState, pub(crate) state: ActivityState,
@@ -493,6 +508,23 @@ impl Default for ActivityState {
} }
fn get_progress_bar(done: u64, expected: u64) -> Option<Cow<'static, str>> { fn get_progress_bar(done: u64, expected: u64) -> Option<Cow<'static, str>> {
if expected == 0 {
return None;
}
// ○◔◑◕●
// ○◎◉●
// ▁▂▃▄▅▆▇█
// ▏▎▍▌▋▊▉█
// ░▒▓█
//
// 🮷 download
// 🮸 upload
// 🮵
// 🮶
get_progress_bar_clockwise_circle(done, expected)
}
fn get_progress_bar_fade_in(done: u64, expected: u64) -> Option<Cow<'static, str>> {
if expected == 0 { if expected == 0 {
return None; return None;
} }
@@ -520,6 +552,36 @@ fn get_progress_bar(done: u64, expected: u64) -> Option<Cow<'static, str>> {
} }
} }
fn get_progress_bar_clockwise_circle(done: u64, expected: u64) -> Option<Cow<'static, str>> {
if expected == 0 {
return None;
}
let percent = done as f32 / expected as f32;
// ○◔◑◕●
// ○◎◉●
// ▁▂▃▄▅▆▇█
// ▏▎▍▌▋▊▉█
// ░▒▓█
//
// 🮷 download
// 🮸 upload
// 🮵
// 🮶
if percent < 0.25 {
Some(Cow::Borrowed(""))
} else if percent < 0.5 {
Some(Cow::Borrowed(""))
} else if percent < 0.75 {
Some(Cow::Borrowed(""))
} else if percent < 1.0 {
Some(Cow::Borrowed(""))
} else if percent >= 1.0 {
Some(Cow::Borrowed(""))
} else {
None
}
}
fn get_progress_text(done: u64, expected: u64) -> Option<Cow<'static, str>> { fn get_progress_text(done: u64, expected: u64) -> Option<Cow<'static, str>> {
Some(Cow::Owned(format!("[{}/{}]", done, expected))) Some(Cow::Owned(format!("[{}/{}]", done, expected)))
} }

View File

@@ -163,6 +163,10 @@ impl ActivityTreeStream {
activity_start_verify_paths.parent, activity_start_verify_paths.parent,
Activity::VerifyPaths(ActivityVerifyPaths { Activity::VerifyPaths(ActivityVerifyPaths {
state: ActivityState::default(), state: ActivityState::default(),
done: 0,
expected: 0,
running: 0,
failed: 0,
}), }),
)?; )?;
} }

View File

@@ -9,6 +9,7 @@ use crate::Result;
use crate::database::db_handle::DbHandle; use crate::database::db_handle::DbHandle;
use super::RunningUpdate; use super::RunningUpdate;
use super::RunningVerify;
use super::running_build::RunningBuild; use super::running_build::RunningBuild;
pub(crate) async fn nixos_build_target<B, F, A, TN>( pub(crate) async fn nixos_build_target<B, F, A, TN>(
@@ -17,6 +18,7 @@ pub(crate) async fn nixos_build_target<B, F, A, TN>(
flake_path: F, flake_path: F,
attr: A, attr: A,
target_name: TN, target_name: TN,
build_id: i64,
) -> Result<()> ) -> Result<()>
where where
B: AsRef<Path>, B: AsRef<Path>,
@@ -55,7 +57,7 @@ where
let child = command.spawn()?; let child = command.spawn()?;
let mut running_build = RunningBuild::new(db_handle)?; let mut running_build = RunningBuild::new(db_handle, build_id)?;
running_build.run_to_completion(child, target_name).await?; running_build.run_to_completion(child, target_name).await?;
Ok(()) Ok(())
@@ -86,3 +88,26 @@ where
Ok(()) 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(())
}

153
src/nix_util/lock_parser.rs Normal file
View File

@@ -0,0 +1,153 @@
use std::collections::BTreeMap;
use std::path::Path;
use serde::Deserialize;
use serde::Serialize;
use crate::Result;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct FlakeLock {
pub(crate) root: String,
pub(crate) version: u8,
pub(crate) nodes: BTreeMap<String, FlakeLockNode>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct FlakeLockNode {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) locked: Option<FlakeLockNodeLocked>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) original: Option<FlakeLockNodeOriginal>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) inputs: Option<BTreeMap<String, FlakeLockInputValue>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) flake: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct FlakeLockNodeLocked {
#[serde(rename = "lastModified")]
pub(crate) last_modified: i64,
#[serde(rename = "narHash")]
pub(crate) nar_hash: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) url: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) owner: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) repo: Option<String>,
#[serde(rename = "ref", default, skip_serializing_if = "Option::is_none")]
pub(crate) git_ref: Option<String>,
pub(crate) rev: String,
#[serde(rename = "revCount", default, skip_serializing_if = "Option::is_none")]
pub(crate) rev_count: Option<i64>,
#[serde(rename = "type")]
pub(crate) repo_type: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "lowercase", deny_unknown_fields)]
pub(crate) enum FlakeLockNodeOriginal {
Github {
owner: String,
repo: String,
#[serde(rename = "ref", default, skip_serializing_if = "Option::is_none")]
git_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
rev: Option<String>,
},
Git {
url: String,
},
Tarball {
url: String,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged, deny_unknown_fields)]
pub(crate) enum FlakeLockInputValue {
String(String),
List(Vec<String>),
}
pub(crate) async fn parse_flake_lock<F>(flake_path: F) -> Result<FlakeLock>
where
F: AsRef<Path>,
{
let lock_file_path = flake_path.as_ref().join("flake.lock");
let contents = tokio::fs::read_to_string(&lock_file_path).await?;
let parsed_flake_lock = serde_json::from_str(&contents)?;
Ok(parsed_flake_lock)
}
impl FlakeLock {
pub(crate) fn get_root_node(&self) -> Result<&FlakeLockNode> {
let root_node = self.nodes.get(&self.root).ok_or("Root node not found.")?;
Ok(root_node)
}
pub(crate) fn get_inputs<'lock>(
&'lock self,
source_node: &'lock FlakeLockNode,
) -> Result<BTreeMap<&'lock String, &'lock FlakeLockNode>> {
let direct_inputs = match &source_node.inputs {
Some(inputs) => inputs,
None => {
return Ok(BTreeMap::new());
}
};
let mut inputs = BTreeMap::new();
for (input_name, input_id_or_path) in direct_inputs.iter() {
inputs.insert(input_name, self.resolve_path(input_id_or_path)?);
}
Ok(inputs)
}
fn resolve_path<'lock>(
&'lock self,
input_id_or_path: &'lock FlakeLockInputValue,
) -> Result<&'lock FlakeLockNode> {
match input_id_or_path {
FlakeLockInputValue::String(input_id) => {
// If the input value is a plain string, then it refers to a node.
let node = self.nodes.get(input_id).ok_or("Input not found.")?;
Ok(node)
}
FlakeLockInputValue::List(items) => {
// If the input value is a list, then it refers to a path to an input.
let mut current_node = self.get_root_node()?;
for step in items {
let child_input_id_or_path = current_node
.inputs
.as_ref()
.ok_or("Node without inputs while walking input tree.")?
.get(step)
.ok_or("Step not found in inputs while walking input tree")?;
current_node = self.resolve_path(child_input_id_or_path)?;
}
Ok(current_node)
}
}
}
}

View File

@@ -2,16 +2,21 @@ mod activity;
mod activity_tree; mod activity_tree;
mod activity_tree_stream; mod activity_tree_stream;
mod high_level; mod high_level;
mod lock_parser;
mod nix_output_stream; mod nix_output_stream;
mod output_stream; mod output_stream;
mod running_build; mod running_build;
mod running_update; mod running_update;
mod running_verify;
mod tree_iter; mod tree_iter;
pub(crate) use activity_tree::ActivityIdAlreadyInTreeError; pub(crate) use activity_tree::ActivityIdAlreadyInTreeError;
pub(crate) use activity_tree::ActivityIdNotInTreeError; pub(crate) use activity_tree::ActivityIdNotInTreeError;
pub(crate) use high_level::*; pub(crate) use high_level::*;
pub(crate) use lock_parser::FlakeLock;
pub(crate) use lock_parser::parse_flake_lock;
pub(crate) use nix_output_stream::NixOutputStream; pub(crate) use nix_output_stream::NixOutputStream;
pub(crate) use output_stream::OutputLine; pub(crate) use output_stream::OutputLine;
pub(crate) use output_stream::OutputLineStream; pub(crate) use output_stream::OutputLineStream;
pub(crate) use running_build::RunningBuild; pub(crate) use running_build::RunningBuild;
pub(crate) use running_update::RunningUpdate; pub(crate) use running_update::RunningUpdate;
pub(crate) use running_verify::RunningVerify;

View File

@@ -21,20 +21,21 @@ use super::nix_output_stream::ActivityResultMessage;
use super::nix_output_stream::NixMessage; use super::nix_output_stream::NixMessage;
use super::tree_iter::DrawDagEntry; use super::tree_iter::DrawDagEntry;
use super::tree_iter::ReverseTreeIter; use super::tree_iter::ReverseTreeIter;
use super::tree_iter::get_draw_order;
pub(crate) struct RunningBuild<'db> { pub(crate) struct RunningBuild<'db> {
db_handle: &'db DbHandle, db_handle: &'db DbHandle,
activity_tree: ActivityTreeStream, activity_tree: ActivityTreeStream,
last_announce: Option<Instant>, last_announce: Option<Instant>,
build_id: i64,
} }
impl<'db> RunningBuild<'db> { impl<'db> RunningBuild<'db> {
pub(crate) fn new(db_handle: &'db DbHandle) -> Result<Self> { pub(crate) fn new(db_handle: &'db DbHandle, build_id: i64) -> Result<Self> {
Ok(RunningBuild { Ok(RunningBuild {
db_handle, db_handle,
activity_tree: ActivityTreeStream::new(), activity_tree: ActivityTreeStream::new(),
last_announce: None, last_announce: None,
build_id,
}) })
} }
@@ -46,16 +47,6 @@ impl<'db> RunningBuild<'db> {
where where
TN: AsRef<str>, TN: AsRef<str>,
{ {
let foo = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
let now = Instant::now();
let build_id: i64 = sqlx::query(
r#"INSERT INTO build (start_time, target) SELECT unixepoch('now'), ? RETURNING id"#,
)
.bind(target_name.as_ref())
.fetch_one(&self.db_handle.conn)
.await?
.try_get("id")?;
let output_stream = OutputStream::from_child(&mut child)?; let output_stream = OutputStream::from_child(&mut child)?;
let mut nix_output_stream: NixOutputStream<OutputStream> = let mut nix_output_stream: NixOutputStream<OutputStream> =
NixOutputStream::new(output_stream); NixOutputStream::new(output_stream);
@@ -82,7 +73,7 @@ impl<'db> RunningBuild<'db> {
.code() .code()
.expect("Process should have an exit code."), .expect("Process should have an exit code."),
) )
.bind(build_id) .bind(self.build_id)
.execute(&self.db_handle.conn) .execute(&self.db_handle.conn)
.await? .await?
.rows_affected(); .rows_affected();
@@ -107,7 +98,12 @@ impl<'db> RunningBuild<'db> {
}; };
match message { match message {
NixAction::Msg(msg_message) => { NixAction::Msg(msg_message) => {
if msg_message.level > 0 && msg_message.level < 5 { if msg_message.level > 0
&& msg_message.level < 5
&& !(msg_message.level == 4 && msg_message.msg.starts_with("linking "))
&& !(msg_message.level == 4 && msg_message.msg.starts_with("evaluating file "))
&& !(msg_message.level == 3 && msg_message.msg.starts_with("checking path "))
{
eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg); eprintln!("LOG MESSAGE {}: {}", msg_message.level, msg_message.msg);
} }
} }
@@ -129,9 +125,16 @@ impl<'db> RunningBuild<'db> {
ActivityResultMessage::SetPhase(_activity_result_set_phase) => { ActivityResultMessage::SetPhase(_activity_result_set_phase) => {
self.print_current_status(); self.print_current_status();
} }
ActivityResultMessage::Progress(_activity_result_progress) => { ActivityResultMessage::Progress(activity_result_progress) => {
if activity_result_progress.done == activity_result_progress.expected
&& activity_result_progress.expected > 0
{
// If we finished the in-progress activity, always render so we show the 100% done version.
self.print_current_status();
} else {
self.maybe_print_current_status(); self.maybe_print_current_status();
} }
}
ActivityResultMessage::SetExpected(_activity_result_set_expected) => { ActivityResultMessage::SetExpected(_activity_result_set_expected) => {
self.maybe_print_current_status(); self.maybe_print_current_status();
} }

View File

@@ -1,5 +1,4 @@
use std::borrow::Cow; use std::borrow::Cow;
use std::ops::Deref;
use std::time::Duration; use std::time::Duration;
use std::time::Instant; use std::time::Instant;
@@ -12,7 +11,6 @@ use crate::nix_util::nix_output_stream::NixAction;
use crate::nix_util::output_stream::OutputStream; use crate::nix_util::output_stream::OutputStream;
use crate::nix_util::tree_iter::ForwardTreeIter; use crate::nix_util::tree_iter::ForwardTreeIter;
use super::activity::Activity;
use super::activity_tree::ActivityTreeEntry; use super::activity_tree::ActivityTreeEntry;
use super::activity_tree_stream::ActivityTreeStream; use super::activity_tree_stream::ActivityTreeStream;
use super::nix_output_stream::ActivityResultMessage; use super::nix_output_stream::ActivityResultMessage;
@@ -49,7 +47,7 @@ impl RunningUpdate {
} }
let exit_status = exit_status_handle.await?; let exit_status = exit_status_handle.await?;
println!("nix build status was: {}", exit_status); println!("nix update status was: {}", exit_status);
Ok(()) Ok(())
} }
@@ -171,25 +169,8 @@ fn is_match_predicate(entry: &ActivityTreeEntry) -> bool {
entry.get_activity().get_progress_text().is_some() entry.get_activity().get_progress_text().is_some()
} }
pub(crate) fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool { fn is_transparent_predicate(entry: &ActivityTreeEntry) -> bool {
return false; return false;
// match entry.get_activity() {
// Activity::Root(_activity_root) => true,
// Activity::Unknown(_activity_unknown) => false,
// Activity::CopyPath(_activity_copy_path) => false,
// Activity::FileTransfer(_activity_file_transfer) => false,
// Activity::Realize(_activity_realize) => true,
// Activity::CopyPaths(_activity_copy_paths) => true,
// Activity::Builds(_activity_builds) => true,
// Activity::Build(_activity_build) => false,
// Activity::OptimizeStore(_activity_optimize_store) => false,
// Activity::VerifyPaths(_activity_verify_paths) => false,
// Activity::Substitute(_activity_substitute) => false,
// Activity::QueryPathInfo(_activity_query_path_info) => false,
// Activity::PostBuildHook(_activity_post_build_hook) => false,
// Activity::BuildWaiting(_activity_build_waiting) => true,
// Activity::FetchTree(_activity_fetch_tree) => false,
// }
} }
fn is_alive_predicate(entry: &ActivityTreeEntry) -> bool { fn is_alive_predicate(entry: &ActivityTreeEntry) -> bool {

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 < 5 {
// 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()
}