file_engine/
progress.rs

1use std::path::PathBuf;
2
3use tokio::sync::mpsc;
4
5use crate::profiler::Entry;
6
7/// See dev-docs/design/handle-progress.md. Discrete per-entry events rather
8/// than a cumulative snapshot; `EntryFailed` carries only the `Entry`,
9/// not the `Error` — `Error` isn't `Clone` (it wraps `std::io::Error`),
10/// and the failure detail is already available from the operation's
11/// final `OperationOutcome.failed` once the handle resolves.
12#[derive(Debug, Clone)]
13pub enum Progress {
14    /// Emitted once per phase, before any entries in that phase start.
15    /// `bytes_total` is `None` for phases with nothing byte-sized to
16    /// report (the delete sweeps). Can be emitted more than once per
17    /// operation — `sync` emits it once per phase (copy, then delete).
18    Started { bytes_total: Option<u64>, entries_total: usize },
19    EntryStarted { entry: Entry },
20    EntryCompleted { entry: Entry },
21    EntryFailed { entry: Entry },
22    /// The directory-creation pre-pass (`operations/pipeline.rs`'s
23    /// `ensure_directories_exist`), separate from `Started`/`Entry*`
24    /// since it operates on `DirEntry`, not `Entry` — carrying only the
25    /// destination path (not the full `DirEntry`) is enough for a
26    /// caller to show "N/M directories created" without extending
27    /// every `Entry`-typed variant to also accept `DirEntry`. Added
28    /// after a real run against a USB-connected exFAT drive spent about
29    /// a minute silently creating ~7,700 directories one at a time
30    /// before `Started` (for the file-copy phase) was ever emitted —
31    /// see dev-docs/design/handle-progress.md.
32    DirectoriesStarted { total: usize },
33    DirectoryCompleted { path: PathBuf },
34    DirectoryFailed { path: PathBuf },
35}
36
37/// `Send + Sync + Clone` sender wrapper threaded into every execution
38/// path that processes entries. Backed by an unbounded channel
39/// specifically so `.send()` is synchronous, not `async` — needed
40/// because `compress.rs`'s writer runs inside `spawn_blocking`, a
41/// non-async context that can't await a bounded channel's backpressure.
42#[derive(Clone)]
43pub(crate) struct ProgressReporter {
44    tx: mpsc::UnboundedSender<Progress>,
45}
46
47impl ProgressReporter {
48    pub(crate) fn new(tx: mpsc::UnboundedSender<Progress>) -> Self {
49        Self { tx }
50    }
51
52    /// A reporter whose receiver is immediately dropped — for tests that
53    /// don't care about progress and don't want to plumb a receiver
54    /// through just to ignore it.
55    #[cfg(test)]
56    pub(crate) fn noop() -> Self {
57        let (tx, _rx) = mpsc::unbounded_channel();
58        Self { tx }
59    }
60
61    /// A closed receiver means nobody's listening (the caller never
62    /// called `.progress()`, or dropped the stream) — not a failure,
63    /// just nothing to report to.
64    pub(crate) fn send(&self, progress: Progress) {
65        let _ = self.tx.send(progress);
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use std::path::PathBuf;
72
73    use super::*;
74
75    fn entry(name: &str) -> Entry {
76        Entry {
77            path: PathBuf::from(name),
78            relative_path: PathBuf::from(name),
79            size: 1,
80            modified: None,
81        }
82    }
83
84    #[test]
85    fn send_after_receiver_dropped_does_not_panic() {
86        let (tx, rx) = mpsc::unbounded_channel();
87        let reporter = ProgressReporter::new(tx);
88        drop(rx);
89
90        reporter.send(Progress::EntryStarted { entry: entry("a") });
91    }
92
93    #[tokio::test]
94    async fn sends_are_received_in_order() {
95        let (tx, mut rx) = mpsc::unbounded_channel();
96        let reporter = ProgressReporter::new(tx);
97
98        reporter.send(Progress::Started { bytes_total: Some(10), entries_total: 2 });
99        reporter.send(Progress::EntryStarted { entry: entry("a") });
100        reporter.send(Progress::EntryCompleted { entry: entry("a") });
101
102        drop(reporter); // otherwise the last recv() below blocks forever
103
104        assert!(matches!(rx.recv().await, Some(Progress::Started { .. })));
105        assert!(matches!(rx.recv().await, Some(Progress::EntryStarted { .. })));
106        assert!(matches!(rx.recv().await, Some(Progress::EntryCompleted { .. })));
107        assert!(rx.recv().await.is_none());
108    }
109}
file_engine/progress.rs

file_engine/
progress.rs
