diff --git a/src/deb/mod.rs b/src/deb/mod.rs index ccde698..0c27e1a 100644 --- a/src/deb/mod.rs +++ b/src/deb/mod.rs @@ -296,6 +296,10 @@ impl BuildView for RecordLogView<'_> { self.inner.progress(label, pos, total); } + fn byte_progress(&self, label: &str, pos: u64, total: u64) { + self.inner.byte_progress(label, pos, total); + } + fn sink(&self) -> Option> { Some(self.log.clone()) } @@ -506,6 +510,7 @@ async fn build_binary_package_impl( series, chroot_arch, mirror.as_deref(), + view, ) { Ok((record, keep)) => { keep_upper = keep; @@ -795,9 +800,32 @@ async fn build_binary_package_impl( result } +/// The staged-tree copy of a fork is bounded (the base's whole tree) +/// and, without reflinks, takes minutes: announce it on the view, then +/// feed byte progress once the fork's watcher has measured the total. +/// Callers hand [`session::CopyProgress`] the box through `as_deref()`. +fn copy_report<'a>( + view: Option<&'a dyn BuildView>, + base_id: &str, +) -> Option> { + view.map(|view| { + let label = format!("Copying the staged tree of record {base_id}"); + view.message(&label); + Box::new(move |pos: u64, total: u64| { + let size = crate::utils::size::format_bytes; + view.byte_progress( + &format!("{label} ({}/{})", size(pos), size(total)), + pos, + total, + ) + }) as Box + }) +} + /// Resolve the record a build runs in, per the `--incremental` request, /// against the real sessions root and the chroot tarball the bootstrap /// would use today. See [`resolve_record_in`]. +#[allow(clippy::too_many_arguments)] fn resolve_record( incremental: &Incremental, host_tree: &Path, @@ -806,6 +834,7 @@ fn resolve_record( series: &str, chroot_arch: Option<&str>, mirror: Option<&str>, + view: &dyn BuildView, ) -> Result<(session::Session, bool), Box> { resolve_record_in( &session::sessions_root(), @@ -814,11 +843,14 @@ fn resolve_record( host_tree, identity, version, + Some(view), ) } /// The testable core of [`resolve_record`] over an explicit sessions root -/// and cached-tarball digest (the fork gate's reference). +/// and cached-tarball digest (the fork gate's reference). `view` is the +/// live build view when there is one; the fork reports its staged-tree +/// copy through it. /// /// Every local build runs in its own record: [`Incremental::Off`] gets a /// brand-new one, [`Incremental::Auto`] and [`Incremental::Id`] fork the @@ -833,6 +865,7 @@ fn resolve_record_in( host_tree: &Path, identity: &session::SessionIdentity, version: &str, + view: Option<&dyn BuildView>, ) -> Result<(session::Session, bool), Box> { match incremental { Incremental::Off => Ok(( @@ -854,6 +887,7 @@ fn resolve_record_in( } else { "no successful record among the last usable ones; the newest record" }; + let report = copy_report(view, &base.manifest().id); match session::fork_record_in( records_root, &base, @@ -861,6 +895,7 @@ fn resolve_record_in( identity, version, cache_tarball_sha256, + report.as_deref(), ) { Ok(record) => { log::info!( @@ -901,6 +936,7 @@ fn resolve_record_in( "Base of the incremental build: record {} (pinned by id)", base.manifest().id ); + let report = copy_report(view, &base.manifest().id); Ok(( session::fork_record_in( records_root, @@ -909,6 +945,7 @@ fn resolve_record_in( identity, version, cache_tarball_sha256, + report.as_deref(), )?, true, )) @@ -1767,6 +1804,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect("a plain run always resolves to a fresh record"); assert!(!keep_upper, "a plain run drops its staged tree"); @@ -1798,6 +1836,7 @@ mod tests { &identity, "1.0", &cache_sha(&tarball), + None, ) .unwrap(); let b_id = b.manifest().id; @@ -1811,6 +1850,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect("the auto selection must fork a candidate record"); assert!(keep_upper); @@ -1836,6 +1876,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect("no candidates must degrade to a fresh build"); assert!(keep_upper, "the first record becomes a base"); @@ -1877,6 +1918,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect("the record carrying a staged tree is still the base"); assert_eq!(record.manifest().base.as_deref(), Some(carrier_id.as_str())); @@ -1907,6 +1949,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect("a pinned failed record is a valid base"); assert!(keep_upper); @@ -1940,6 +1983,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect_err("records of other trees must not be forked by id"); assert!( @@ -1981,6 +2025,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect_err("an ambiguous prefix must be refused"); assert!( @@ -2029,6 +2074,7 @@ mod tests { &host_tree, &identity, "1.0", + None, ) .expect_err("a foreign-identity record must not be forked"); assert!( diff --git a/src/deb/session.rs b/src/deb/session.rs index 5c526cf..db13e11 100644 --- a/src/deb/session.rs +++ b/src/deb/session.rs @@ -1062,6 +1062,63 @@ fn copy_dir(src: &Path, dst: &Path) -> io::Result<()> { } } +/// Byte progress of a staged-tree copy: the bytes copied so far and the +/// source's measured total. Best-effort — `cp` has no progress output to +/// tap, so the destination is re-measured in the background while it +/// runs, and the completed total is reported when the copy returns. +pub type CopyProgress<'a> = &'a (dyn Fn(u64, u64) + Sync); + +/// [`copy_dir`], reporting byte progress through `progress` when given. +/// A staged-tree copy is bounded (the base's whole tree) and, without +/// reflinks, takes minutes: the source is sized up front (the copy's +/// total) and a thread re-measures the growing destination while `cp` +/// runs. Unmeasurable or empty sources copy unreported. +fn copy_dir_with_progress( + src: &Path, + dst: &Path, + progress: Option>, +) -> io::Result<()> { + use std::sync::atomic::{AtomicBool, Ordering}; + + let Some(progress) = progress else { + return copy_dir(src, dst); + }; + let total = crate::utils::size::tree_sizes(&[src]).pop().unwrap_or(0); + if total == 0 { + return copy_dir(src, dst); + } + progress(0, total); + + let stop = Arc::new(AtomicBool::new(false)); + let stopped = stop.clone(); + let watched = dst.to_path_buf(); + let result = std::thread::scope(|scope| { + scope.spawn(|| { + // Every measure walks what cp wrote so far: back off rather + // than compete with the copy. + let mut pause = Duration::from_millis(500); + while !stopped.load(Ordering::SeqCst) { + std::thread::sleep(pause); + if stopped.load(Ordering::SeqCst) { + return; + } + let copied = crate::utils::size::tree_sizes(&[watched.as_path()]) + .pop() + .unwrap_or(0); + progress(copied.min(total), total); + pause = (pause * 2).min(Duration::from_secs(2)); + } + }); + let result = copy_dir(src, dst); + stop.store(true, Ordering::SeqCst); + result + }); + // The scope joined the watcher: the completion report cannot + // interleave with a background one. + progress(total, total); + result +} + /// Fork `base` into a new record of `identity` under the sessions root. /// See [`fork_record_in`]. pub fn fork_record( @@ -1070,6 +1127,7 @@ pub fn fork_record( identity: &SessionIdentity, version: &str, cache_tarball_sha256: &str, + progress: Option>, ) -> Result { fork_record_in( &sessions_root(), @@ -1078,6 +1136,7 @@ pub fn fork_record( identity, version, cache_tarball_sha256, + progress, ) } @@ -1099,6 +1158,7 @@ pub(crate) fn fork_record_in( identity: &SessionIdentity, version: &str, cache_tarball_sha256: &str, + progress: Option>, ) -> Result { let base_manifest = base.manifest(); if base_manifest.identity != *identity { @@ -1157,13 +1217,14 @@ pub(crate) fn fork_record_in( })?; if carry_upper { - let copied = copy_dir(&base.upper_dir(), &record.upper_dir()).and_then(|_bytes| { - if base.snapshot_file().is_file() { - fs::copy(base.snapshot_file(), record.snapshot_file()) - } else { - Ok(0) - } - }); + let copied = copy_dir_with_progress(&base.upper_dir(), &record.upper_dir(), progress) + .and_then(|_bytes| { + if base.snapshot_file().is_file() { + fs::copy(base.snapshot_file(), record.snapshot_file()) + } else { + Ok(0) + } + }); if let Err(e) = copied { // The partial fork is inert (its manifest is written but // its staged tree is incomplete): remove it rather than @@ -2822,7 +2883,8 @@ mod tests { // Everything matches: the staged tree, its snapshot and the tree // journal are carried. - let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", &sha).unwrap(); + let fork = + fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", &sha, None).unwrap(); assert_eq!(fork.manifest().base.as_deref(), Some(base_id.as_str())); assert_eq!( fs::read_to_string(fork.upper_dir().join("src/main.o")).unwrap(), @@ -2833,7 +2895,8 @@ mod tests { drop(fork); // The changelog version changed: cold fork. - let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.1", &sha).unwrap(); + let fork = + fork_record_in(root.path(), &base, &host_tree, &identity, "1.1", &sha, None).unwrap(); assert_eq!(fork.manifest().tree_version, "1.1"); assert!(!fork.upper_dir().join("src/main.o").exists()); assert!(!fork.snapshot_file().exists()); @@ -2841,13 +2904,22 @@ mod tests { drop(fork); // The toolchain moved (the cached tarball changed): cold fork. - let fork = - fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", "deadbeef").unwrap(); + let fork = fork_record_in( + root.path(), + &base, + &host_tree, + &identity, + "1.0", + "deadbeef", + None, + ) + .unwrap(); assert!(!fork.upper_dir().join("src/main.o").exists()); drop(fork); // No cached tarball yet (the sha is unknown): cold fork. - let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", "").unwrap(); + let fork = + fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", "", None).unwrap(); assert!(!fork.upper_dir().join("src/main.o").exists()); } @@ -2862,7 +2934,8 @@ mod tests { let base = record_with_upper(root.path(), &tarball, &host_tree, &identity, "1.0", None); let base_upper = base.upper_dir().join("src/main.o"); - let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", &sha).unwrap(); + let fork = + fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", &sha, None).unwrap(); fs::write(fork.upper_dir().join("src/main.o"), b"changed").unwrap(); fs::write(fork.snapshot_file(), "mutated\n").unwrap(); assert_eq!(fs::read_to_string(&base_upper).unwrap(), "objects"); @@ -2872,6 +2945,47 @@ mod tests { ); } + /// A fork with a copy reporter opens with the source's measured total + /// and closes with the completed copy — the view renders the progress + /// bar from exactly these bounds. + #[test] + fn fork_reports_the_copy_progress() { + let (root, tarball) = sessions_fixture(); + let identity = test_identity(); + let host_tree = root.path().join("tree"); + let sha = sha256_file(&tarball).unwrap(); + let base = record_with_upper( + root.path(), + &tarball, + &host_tree, + &identity, + "1.0", + Some(OUTCOME_SUCCESS), + ); + let total = crate::utils::size::tree_sizes(&[base.upper_dir().as_path()]).remove(0); + assert!(total > 0); + + let events = Mutex::new(Vec::new()); + let fork = fork_record_in( + root.path(), + &base, + &host_tree, + &identity, + "1.0", + &sha, + Some(&|pos, tot| events.lock().unwrap().push((pos, tot))), + ) + .unwrap(); + assert_eq!( + fs::read_to_string(fork.upper_dir().join("src/main.o")).unwrap(), + "objects" + ); + + let events = events.into_inner().unwrap(); + assert_eq!(events.first(), Some(&(0, total))); + assert_eq!(events.last(), Some(&(total, total))); + } + /// The user scenario: record A succeeds, a fork B of it fails — the /// default selection must pick A (the newest success), not B (the /// newest record); with no success at all it takes the newest. @@ -2890,7 +3004,7 @@ mod tests { Some(OUTCOME_SUCCESS), ); let a_id = a.manifest().id; - let b = fork_record_in(root.path(), &a, &host_tree, &identity, "1.0", &sha).unwrap(); + let b = fork_record_in(root.path(), &a, &host_tree, &identity, "1.0", &sha, None).unwrap(); let b_id = b.manifest().id; b.set_outcome(OUTCOME_FAILED); drop((a, b)); diff --git a/src/report.rs b/src/report.rs index e545129..f29adf4 100644 --- a/src/report.rs +++ b/src/report.rs @@ -92,6 +92,12 @@ pub trait BuildView: Send + Sync { /// retrieval); `pos` runs from 0 to `total`. fn progress(&self, _label: &str, _pos: usize, _total: usize) {} + /// Determinate progress of a byte transfer within the current phase + /// (e.g. the staged-tree copy forked from an incremental base): + /// `pos` of `total` bytes. Like [`BuildView::progress`], but adapters + /// render byte units for it. + fn byte_progress(&self, _label: &str, _pos: u64, _total: u64) {} + /// Sink receiving every raw subprocess line while a build command runs, /// when this view consumes the lines itself (live rewriting, tee to a /// log file, forwarding over the network). `None` lets the caller fall diff --git a/src/ui/deb.rs b/src/ui/deb.rs index b05a320..7fa7735 100644 --- a/src/ui/deb.rs +++ b/src/ui/deb.rs @@ -46,6 +46,9 @@ struct Pipeline { errors: Vec, last_draw: Instant, bar_total: u64, + /// Whether the bar's counters are bytes ([`BuildView::byte_progress`]) + /// rather than plain counts + bar_bytes: bool, } /// State shared between [`DebUi`] and its sinks @@ -119,6 +122,7 @@ impl DebUi { errors: Vec::new(), last_draw: Instant::now(), bar_total: 0, + bar_bytes: false, }), tee: Mutex::new(None), log_path: Mutex::new(log_path), @@ -195,6 +199,7 @@ impl DebUi { st.classifier = classifier; st.lines.clear(); st.bar_total = 0; + st.bar_bytes = false; st.last_draw = Instant::now(); } if self.active() { @@ -309,6 +314,19 @@ impl crate::report::BuildView for DebUi { self.shared.top.set_message(label.to_string()); } + fn byte_progress(&self, label: &str, pos: u64, total: u64) { + if !self.active() || total == 0 { + return; + } + apply_byte_progress( + &self.shared.top, + &mut self.shared.state.lock().unwrap(), + pos, + total, + ); + self.shared.top.set_message(label.to_string()); + } + fn sink(&self) -> Option> { Some(Arc::new(Sink { shared: self.shared.clone(), @@ -384,7 +402,8 @@ fn apply_progress(top: &ProgressBar, st: &mut Pipeline, pos: u64, total: u64) { if total == 0 { return; } - if st.bar_total != total { + if st.bar_bytes || st.bar_total != total { + st.bar_bytes = false; st.bar_total = total; top.set_style(determinate_style()); top.set_length(total); @@ -392,6 +411,21 @@ fn apply_progress(top: &ProgressBar, st: &mut Pipeline, pos: u64, total: u64) { top.set_position(pos.min(total)); } +/// Update the determinate bar with byte counters (human-formatted by the +/// template), restyling it when needed +fn apply_byte_progress(top: &ProgressBar, st: &mut Pipeline, pos: u64, total: u64) { + if total == 0 { + return; + } + if !st.bar_bytes || st.bar_total != total { + st.bar_bytes = true; + st.bar_total = total; + top.set_style(byte_style()); + top.set_length(total); + } + top.set_position(pos.min(total)); +} + /// Push a line into the rolling pane and redraw (throttled) /// /// When the pane is full, the oldest non-error line is dropped first so @@ -530,6 +564,15 @@ fn determinate_style() -> ProgressStyle { .progress_chars("=> ") } +/// Status bar style with a determinate progress bar over a byte +/// transfer: the counters render as byte units (`{bytes}`/`{total_bytes}`) +fn byte_style() -> ProgressStyle { + ProgressStyle::default_bar() + .template("> {spinner:.blue} {prefix}\n {msg} [{bar:24.cyan/blue}] {bytes}/{total_bytes}") + .expect("valid template") + .progress_chars("=> ") +} + /// Whether stdout is a terminal fn is_stdout_tty() -> bool { // SAFETY: isatty only inspects the file descriptor @@ -722,6 +765,7 @@ mod tests { errors: Vec::new(), last_draw: Instant::now(), bar_total: 0, + bar_bytes: false, }), tee: Mutex::new(None), log_path: Mutex::new(std::env::temp_dir().join("pkh-pane-test.log")), @@ -764,6 +808,7 @@ mod tests { errors: Vec::new(), last_draw: Instant::now(), bar_total: 0, + bar_bytes: false, }), tee: Mutex::new(None), log_path: Mutex::new(std::env::temp_dir().join("pkh-interrupt-report-test.log")),