gravityfile_ops/
progress.rs

1//! Progress reporting types for file operations.
2
3use std::path::PathBuf;
4use std::time::{Duration, Instant};
5
6use serde::{Deserialize, Serialize};
7
8use crate::OperationError;
9
10const MAX_DIAGNOSTIC_PAYLOAD_BYTES: usize = 8 * 1024;
11
12/// The type of operation being performed.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
14pub enum OperationType {
15    Copy,
16    Move,
17    Delete,
18    /// Items restored from the system Trash.
19    Restore,
20    Rename,
21    CreateFile,
22    CreateDirectory,
23}
24
25impl std::fmt::Display for OperationType {
26    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        match self {
28            Self::Copy => write!(f, "Copy"),
29            Self::Move => write!(f, "Move"),
30            Self::Delete => write!(f, "Delete"),
31            Self::Restore => write!(f, "Restore"),
32            Self::Rename => write!(f, "Rename"),
33            Self::CreateFile => write!(f, "Create file"),
34            Self::CreateDirectory => write!(f, "Create directory"),
35        }
36    }
37}
38
39/// Progress information for an ongoing operation.
40#[derive(Debug, Clone)]
41pub struct OperationProgress {
42    /// The type of operation.
43    pub operation_type: OperationType,
44    /// Number of files/directories completed.
45    pub files_completed: usize,
46    /// Total number of files/directories to process.
47    pub files_total: usize,
48    /// Number of bytes processed so far.
49    pub bytes_processed: u64,
50    /// Total bytes to process (may be 0 if unknown).
51    pub bytes_total: u64,
52    /// The file currently being processed.
53    pub current_file: Option<PathBuf>,
54    /// Errors encountered so far.
55    pub errors: Vec<OperationError>,
56    /// When the operation started; the basis for throughput and ETA.
57    pub started_at: Instant,
58}
59
60impl OperationProgress {
61    /// Create a new progress tracker for an operation.
62    pub fn new(operation_type: OperationType, files_total: usize, bytes_total: u64) -> Self {
63        Self {
64            operation_type,
65            files_completed: 0,
66            files_total,
67            bytes_processed: 0,
68            bytes_total,
69            current_file: None,
70            errors: Vec::new(),
71            started_at: Instant::now(),
72        }
73    }
74
75    /// Average bytes per second since the operation started.
76    pub fn bytes_per_second(&self) -> f64 {
77        let elapsed = self.started_at.elapsed().as_secs_f64();
78        if elapsed > 0.0 {
79            self.bytes_processed as f64 / elapsed
80        } else {
81            0.0
82        }
83    }
84
85    /// Estimated time remaining, when the byte total and rate are known.
86    pub fn eta(&self) -> Option<Duration> {
87        let rate = self.bytes_per_second();
88        let remaining = self.bytes_total.checked_sub(self.bytes_processed)?;
89        (rate >= 1.0 && self.bytes_total > 0)
90            .then(|| Duration::from_secs_f64(remaining as f64 / rate))
91    }
92
93    /// Record bytes transferred within the current item.
94    pub fn add_bytes(&mut self, bytes: u64) {
95        self.bytes_processed = self.bytes_processed.saturating_add(bytes);
96    }
97
98    /// Get the estimated progress as a percentage (0.0 to 100.0).
99    /// Totals may become stale; 100% is not a terminal success signal.
100    pub fn percentage(&self) -> f64 {
101        if self.bytes_total > 0 {
102            ((self.bytes_processed as f64 / self.bytes_total as f64) * 100.0).min(100.0)
103        } else if self.files_total > 0 {
104            ((self.files_completed as f64 / self.files_total as f64) * 100.0).min(100.0)
105        } else {
106            0.0
107        }
108    }
109
110    /// Check if the operation has any errors.
111    pub fn has_errors(&self) -> bool {
112        !self.errors.is_empty()
113    }
114
115    /// Get the number of errors.
116    pub fn error_count(&self) -> usize {
117        self.errors.iter().fold(0usize, |total, error| {
118            total.saturating_add(error.represented_count())
119        })
120    }
121
122    /// Retain at most 1,000 diagnostic records, preserving total counts through
123    /// an explicit omission record after the first 999 details when truncated.
124    /// Individual path/message payloads above 8 KiB become omission records too;
125    /// native paths are never shortened into misleading partial filenames.
126    pub fn add_error(&mut self, error: OperationError) {
127        const MAX_RETAINED_ERRORS: usize = 1000;
128        let payload_bytes = error
129            .path
130            .as_os_str()
131            .as_encoded_bytes()
132            .len()
133            .saturating_add(error.message.len());
134        let error = if payload_bytes > MAX_DIAGNOSTIC_PAYLOAD_BYTES {
135            let mut omitted = OperationError::omitted(error.represented_count());
136            omitted
137                .message
138                .push_str(" (diagnostic payload exceeded 8 KiB)");
139            omitted
140        } else {
141            error
142        };
143        if self.errors.len() < MAX_RETAINED_ERRORS {
144            self.errors.push(error);
145        } else {
146            let omitted = self
147                .errors
148                .drain(MAX_RETAINED_ERRORS - 1..)
149                .fold(error.represented_count(), |total, prior| {
150                    total.saturating_add(prior.represented_count())
151                });
152            self.errors.push(OperationError::omitted(omitted));
153        }
154    }
155
156    /// Update the current file being processed.
157    pub fn set_current_file(&mut self, path: Option<PathBuf>) {
158        self.current_file = path;
159    }
160
161    /// Increment the completed count and add bytes, saturating at counter limits.
162    pub fn complete_file(&mut self, bytes: u64) {
163        self.files_completed = self.files_completed.saturating_add(1);
164        self.bytes_processed = self.bytes_processed.saturating_add(bytes);
165    }
166}
167
168/// Result of a completed operation.
169#[derive(Debug, Clone)]
170pub struct OperationComplete {
171    /// Whether cancellation or an explicit abort stopped the operation.
172    /// Completed items remain reflected in the counts and undo records.
173    pub cancelled: bool,
174    /// The type of operation.
175    pub operation_type: OperationType,
176    /// Number of items successfully processed.
177    pub succeeded: usize,
178    /// Number of items that failed.
179    pub failed: usize,
180    /// Number of items the user chose to skip. Skips are not failures.
181    pub skipped: usize,
182    /// Total bytes processed.
183    pub bytes_processed: u64,
184    /// Errors that occurred.
185    pub errors: Vec<OperationError>,
186}
187
188impl OperationComplete {
189    /// Check if the operation was fully successful.
190    pub fn is_success(&self) -> bool {
191        self.failed == 0 && self.errors.is_empty() && !self.cancelled
192    }
193
194    /// Get a human-readable summary of the operation.
195    pub fn summary(&self) -> String {
196        let action = match self.operation_type {
197            OperationType::Copy => "Copied",
198            OperationType::Move => "Moved",
199            OperationType::Delete => "Deleted",
200            OperationType::Restore => "Restored",
201            OperationType::Rename => "Renamed",
202            OperationType::CreateFile => "Created",
203            OperationType::CreateDirectory => "Created",
204        };
205
206        let mut summary = format!("{action} {} items", self.succeeded);
207        if self.skipped > 0 {
208            summary.push_str(&format!(", {} skipped", self.skipped));
209        }
210        if self.failed > 0 {
211            summary.push_str(&format!(", {} failed", self.failed));
212        } else if !self.errors.is_empty() {
213            summary.push_str(&format!(
214                ", {} errors",
215                self.errors.iter().fold(0usize, |total, error| total
216                    .saturating_add(error.represented_count()))
217            ));
218        }
219        if self.cancelled {
220            summary.push_str("; operation cancelled");
221        }
222        summary
223    }
224}
225
226/// Progress is replaceable telemetry; a slow consumer must not suspend mutation
227/// cleanup or cancellation. Reserve capacity before constructing the snapshot so
228/// dropped updates never clone potentially large error lists. Terminal results
229/// and conflict requests use reliable sends.
230pub(crate) fn publish_progress<T>(
231    tx: &tokio::sync::mpsc::Sender<T>,
232    make_progress: impl FnOnce() -> T,
233    token: &tokio_util::sync::CancellationToken,
234) {
235    match tx.try_reserve() {
236        Ok(permit) => permit.send(make_progress()),
237        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => token.cancel(),
238        Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {}
239    }
240}
241
242#[cfg(test)]
243mod delivery_tests {
244    use super::*;
245    use tokio_util::sync::CancellationToken;
246
247    #[test]
248    fn diagnostic_payload_limit_preserves_boundaries_and_failure_counts() {
249        let mut progress = OperationProgress::new(OperationType::Copy, 0, 0);
250        let path = PathBuf::from("item");
251        let message = "x".repeat(MAX_DIAGNOSTIC_PAYLOAD_BYTES - 4);
252        progress.add_error(OperationError::new(path.clone(), message.clone()));
253        assert_eq!(progress.errors[0].path, path);
254        assert_eq!(progress.errors[0].message, message);
255        progress.add_error(OperationError::new(path, message + "x"));
256        assert_eq!(progress.errors[1].omitted_count, 1);
257        assert!(progress.errors[1].path.as_os_str().is_empty());
258        assert!(progress.errors[1].message.contains("exceeded 8 KiB"));
259        let mut oversized_omission = OperationError::omitted(37);
260        oversized_omission.message = "界".repeat(MAX_DIAGNOSTIC_PAYLOAD_BYTES);
261        progress.add_error(oversized_omission);
262        assert_eq!(progress.error_count(), 39);
263        assert_eq!(progress.errors[2].omitted_count, 37);
264        for _ in 0..2000 {
265            progress.add_error(OperationError::new(
266                PathBuf::from("p".repeat(MAX_DIAGNOSTIC_PAYLOAD_BYTES + 1)),
267                "failed",
268            ));
269        }
270        assert_eq!(progress.error_count(), 2039);
271        assert_eq!(progress.errors.len(), 1000);
272        assert!(progress.errors.iter().all(|error| {
273            error.path.as_os_str().as_encoded_bytes().len() + error.message.len()
274                <= MAX_DIAGNOSTIC_PAYLOAD_BYTES
275        }));
276    }
277
278    #[cfg(unix)]
279    #[test]
280    fn diagnostic_payload_budget_uses_native_path_bytes() {
281        use std::os::unix::ffi::OsStringExt;
282        let path = PathBuf::from(std::ffi::OsString::from_vec(vec![0xff; 4096]));
283        let mut progress = OperationProgress::new(OperationType::Copy, 0, 0);
284        progress.add_error(OperationError::new(path.clone(), "x".repeat(4096)));
285        assert_eq!(progress.errors[0].path, path);
286        assert_eq!(progress.errors[0].omitted_count, 0);
287        progress.add_error(OperationError::new(path, "x".repeat(4097)));
288        assert_eq!(progress.errors[1].omitted_count, 1);
289    }
290
291    #[test]
292    fn retained_diagnostics_are_bounded_without_losing_total_counts() {
293        let mut progress = OperationProgress::new(OperationType::Copy, 100_000, 0);
294        for index in 0..100_000 {
295            progress.add_error(OperationError::new(
296                PathBuf::from(format!("item-{index}")),
297                "failed",
298            ));
299            assert!(progress.errors.len() <= 1000);
300            if [998, 999, 1000, 99_999].contains(&index) {
301                assert_eq!(progress.error_count(), index + 1);
302            }
303        }
304        assert_eq!(progress.errors.len(), 1000);
305        for (index, error) in progress.errors[..999].iter().enumerate() {
306            assert_eq!(error.path, PathBuf::from(format!("item-{index}")));
307            assert_eq!(error.omitted_count, 0);
308        }
309        assert_eq!(progress.errors[999].omitted_count, 99_001);
310        assert!(progress.errors[999].path.as_os_str().is_empty());
311        assert!(progress.errors[999].message.contains("99001"));
312        assert_eq!(progress.clone().error_count(), 100_000);
313        let completion = OperationComplete {
314            operation_type: OperationType::Copy,
315            cancelled: false,
316            succeeded: 0,
317            failed: 0,
318            skipped: 0,
319            bytes_processed: 0,
320            errors: progress.errors,
321        };
322        assert!(!completion.is_success());
323        assert!(completion.summary().contains("100000 errors"));
324        let legacy: OperationError =
325            serde_json::from_str(r#"{"path":"legacy","message":"failure"}"#).unwrap();
326        assert_eq!(legacy.omitted_count, 0);
327        assert_eq!(legacy.represented_count(), 1);
328    }
329
330    #[test]
331    fn completion_diagnostics_prevent_success_even_without_failure_counts() {
332        for cancelled in [false, true] {
333            for failed in [0, 1] {
334                for errors in [
335                    Vec::new(),
336                    vec![OperationError::new(PathBuf::from("item"), "failure")],
337                ] {
338                    let complete = OperationComplete {
339                        operation_type: OperationType::Copy,
340                        cancelled,
341                        failed,
342                        succeeded: 1,
343                        skipped: 2,
344                        bytes_processed: 0,
345                        errors,
346                    };
347                    assert_eq!(
348                        complete.is_success(),
349                        !cancelled && failed == 0 && complete.errors.is_empty()
350                    );
351                    let summary = complete.summary();
352                    assert!(summary.contains("2 skipped"));
353                    assert_eq!(summary.contains("1 failed"), failed == 1);
354                    assert_eq!(
355                        summary.contains("1 errors"),
356                        failed == 0 && !complete.errors.is_empty()
357                    );
358                    assert_eq!(summary.contains("cancelled"), cancelled);
359                }
360            }
361        }
362    }
363
364    #[test]
365    fn progress_is_bounded_for_unknown_stale_and_extreme_totals() {
366        let mut progress = OperationProgress::new(OperationType::Copy, 0, 0);
367        assert_eq!(progress.percentage(), 0.0);
368        progress.files_total = 4;
369        progress.complete_file(0);
370        assert_eq!(progress.percentage(), 25.0);
371        progress.bytes_total = 8;
372        progress.bytes_processed = 4;
373        assert_eq!(progress.percentage(), 50.0);
374        for files_total in [0, 1, usize::MAX] {
375            for bytes_total in [0, 1, u64::MAX] {
376                progress.files_total = files_total;
377                progress.bytes_total = bytes_total;
378                progress.files_completed = usize::MAX;
379                progress.bytes_processed = u64::MAX;
380                progress.complete_file(u64::MAX);
381                assert_eq!(progress.files_completed, usize::MAX);
382                assert_eq!(progress.bytes_processed, u64::MAX);
383                let expected = if files_total == 0 && bytes_total == 0 {
384                    0.0
385                } else {
386                    100.0
387                };
388                assert_eq!(progress.percentage(), expected);
389            }
390        }
391    }
392
393    #[test]
394    fn full_progress_queue_is_nonblocking_and_does_not_cancel() {
395        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
396        tx.try_send(1).unwrap();
397        let token = CancellationToken::new();
398        publish_progress(
399            &tx,
400            || panic!("full queue must not construct a snapshot"),
401            &token,
402        );
403        assert!(!token.is_cancelled());
404        assert_eq!(rx.try_recv().unwrap(), 1);
405        assert!(rx.try_recv().is_err());
406    }
407
408    #[test]
409    fn closed_progress_receiver_cancels_work() {
410        let (tx, rx) = tokio::sync::mpsc::channel::<usize>(1);
411        drop(rx);
412        let token = CancellationToken::new();
413        publish_progress(
414            &tx,
415            || panic!("closed queue must not construct a snapshot"),
416            &token,
417        );
418        assert!(token.is_cancelled());
419    }
420
421    #[test]
422    fn snapshots_resume_after_backpressure_without_building_discarded_updates() {
423        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
424        let token = CancellationToken::new();
425        let constructed = std::cell::Cell::new(0);
426        let snapshot = || {
427            constructed.set(constructed.get() + 1);
428            constructed.get()
429        };
430        publish_progress(&tx, snapshot, &token);
431        for _ in 0..10_000 {
432            publish_progress(&tx, snapshot, &token);
433        }
434        assert_eq!(constructed.get(), 1);
435        assert_eq!(rx.try_recv().unwrap(), 1);
436        publish_progress(&tx, snapshot, &token);
437        assert_eq!(constructed.get(), 2);
438        assert_eq!(rx.try_recv().unwrap(), 2);
439        assert!(!token.is_cancelled());
440    }
441}
gravityfile_ops/progress.rs

gravityfile_ops/
progress.rs
