dvs/
operation.rs

1//! Operation context (observer hooks, cancel flag), counts, byte totals and
2//! timing helpers.
3
4use std::path::Path;
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
8
9use serde::Serialize;
10
11use crate::files::add::AddResult;
12use crate::files::get::GetRow;
13use crate::files::status::StatusFacts;
14use crate::progress::FileProgress;
15
16/// A shared "stop starting new files" flag. Cloning shares the flag, so a
17/// signal handler can hold one clone while the operation checks another.
18///
19/// Cancelling never interrupts a file already in flight: it finishes
20/// atomically, and only files that had not started are reported `cancelled`.
21#[derive(Debug, Clone, Default)]
22pub struct CancelToken(Arc<AtomicBool>);
23
24impl CancelToken {
25    pub fn new() -> Self {
26        Self::default()
27    }
28
29    pub fn cancel(&self) {
30        self.0.store(true, Ordering::SeqCst);
31    }
32
33    pub fn is_cancelled(&self) -> bool {
34        self.0.load(Ordering::SeqCst)
35    }
36}
37
38/// How one file of an add or status run ended.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum RowState {
41    Succeeded,
42    Failed,
43    /// A cancel stopped it before it started
44    Cancelled,
45}
46
47/// A result row that can be counted into [`Counts`].
48pub trait OperationRow {
49    fn state(&self) -> RowState;
50}
51
52/// How an add or status run ended.
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
54#[serde(rename_all = "snake_case")]
55pub enum RunOutcome {
56    Complete,
57    Partial,
58    Failed,
59    /// A cancel came before the run finished, even if every file had started
60    Cancelled,
61}
62
63impl RunOutcome {
64    /// `cancelled`: the cancel flag was set when the run finished.
65    pub fn of(counts: &Counts, cancelled: bool) -> Self {
66        if cancelled {
67            RunOutcome::Cancelled
68        } else if counts.failed == 0 {
69            RunOutcome::Complete
70        } else if counts.failed == counts.total {
71            RunOutcome::Failed
72        } else {
73            RunOutcome::Partial
74        }
75    }
76}
77
78/// What an operation is about to do, known once selection and storage checks
79/// are done. `status` reports zero `*_to_transfer`.
80#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
81pub struct OperationPlan {
82    pub files_total: u64,
83    pub bytes_total: u64,
84    pub files_to_transfer: u64,
85    pub bytes_to_transfer: u64,
86}
87
88/// Byte totals for an `add` or `get` result: `to_transfer + already_stored
89/// == total`; `transferred` is what actually moved. Rows of unknown size count
90/// in none.
91#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
92pub struct ByteTotals {
93    pub total: u64,
94    pub to_transfer: u64,
95    pub already_stored: u64,
96    pub transferred: u64,
97}
98
99impl ByteTotals {
100    pub(crate) fn count(&mut self, size: u64, stored: bool, transferred: bool) {
101        self.total += size;
102        if stored {
103            self.already_stored += size;
104        } else {
105            self.to_transfer += size;
106        }
107        if transferred {
108            self.transferred += size;
109        }
110    }
111}
112
113/// Row counts for a result: `succeeded + failed + cancelled == total`.
114#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
115pub struct Counts {
116    pub total: usize,
117    pub succeeded: usize,
118    pub failed: usize,
119    pub cancelled: usize,
120}
121
122impl Counts {
123    /// For results that cannot be cancelled.
124    pub fn new(total: usize, failed: usize) -> Self {
125        Counts {
126            total,
127            succeeded: total - failed,
128            failed,
129            cancelled: 0,
130        }
131    }
132
133    pub fn of<'a, R: OperationRow + 'a>(rows: impl IntoIterator<Item = &'a R>) -> Self {
134        let mut counts = Counts::default();
135        for row in rows {
136            counts.total += 1;
137            match row.state() {
138                RowState::Succeeded => counts.succeeded += 1,
139                RowState::Failed => counts.failed += 1,
140                RowState::Cancelled => counts.cancelled += 1,
141            }
142        }
143        counts
144    }
145}
146
147/// A file's final row, in the same shape the command's result will hold.
148#[derive(Debug, Clone, Copy)]
149pub enum FileRow<'a> {
150    Add(&'a AddResult),
151    Get(&'a GetRow),
152    Status(&'a StatusFacts),
153}
154
155/// Hooks an operation calls as it runs. Called from worker threads, possibly
156/// concurrently; implementations serialize their own output.
157pub trait OperationObserver: Send + Sync {
158    /// Once, after selection and storage checks, before any transfer.
159    fn planned(&self, _plan: &OperationPlan) {}
160
161    /// The first file that needs hashing is about to be hashed. Not called
162    /// when nothing is hashed.
163    fn hashing_started(&self) {}
164
165    /// Hashing is done, once for every `hashing_started`, error exits included.
166    fn hashing_finished(&self) {}
167
168    /// A transfer of `total_bytes` for `path` is starting. The returned
169    /// callbacks receive byte deltas and the transfer's success; `None` means
170    /// nobody is watching this transfer.
171    fn transfer_started(&self, _path: &Path, _total_bytes: u64) -> Option<FileProgress> {
172        None
173    }
174
175    /// A file's outcome is final, in completion order.
176    fn file_finished(&self, _row: FileRow<'_>) {}
177}
178
179/// Hashing that starts the first time a file is hashed, from any thread, and
180/// is never reported if none is. Finished on drop if it started, so it is
181/// finished on every exit, `?` and `bail!` included.
182pub(crate) struct Hashing<'a> {
183    observer: &'a dyn OperationObserver,
184    started: std::sync::Once,
185}
186
187impl<'a> Hashing<'a> {
188    pub(crate) fn new(observer: &'a dyn OperationObserver) -> Self {
189        Self {
190            observer,
191            started: std::sync::Once::new(),
192        }
193    }
194
195    /// Call before hashing each file. The first call reports
196    /// `hashing_started`; concurrent callers wait for it, so no hashed file
197    /// is reported before hashing is.
198    pub(crate) fn begin(&self) {
199        self.started.call_once(|| self.observer.hashing_started());
200    }
201
202    /// Finishes hashing here if it started; the same as dropping it.
203    pub(crate) fn finish(self) {}
204}
205
206impl Drop for Hashing<'_> {
207    fn drop(&mut self) {
208        if self.started.is_completed() {
209            self.observer.hashing_finished();
210        }
211    }
212}
213
214/// Hears nothing: what an unobserved operation reports to.
215struct NoopObserver;
216
217impl OperationObserver for NoopObserver {}
218
219/// Adapts an `on_file_start` callback (the pre-observer progress API) into an
220/// observer that only watches transfers.
221pub struct TransferObserver<F>(pub F);
222
223impl<F> OperationObserver for TransferObserver<F>
224where
225    F: Fn(&Path, u64) -> FileProgress + Send + Sync,
226{
227    fn transfer_started(&self, path: &Path, total_bytes: u64) -> Option<FileProgress> {
228        Some((self.0)(path, total_bytes))
229    }
230}
231
232/// Passed to every operation: who, if anyone, observes it, and its cancel flag.
233#[derive(Clone, Default)]
234pub struct OperationContext {
235    observer: Option<Arc<dyn OperationObserver>>,
236    cancel: CancelToken,
237}
238
239impl OperationContext {
240    /// Observed by nobody, never cancelled unless asked.
241    pub fn new() -> Self {
242        Self::default()
243    }
244
245    pub fn with_cancel(mut self, cancel: CancelToken) -> Self {
246        self.cancel = cancel;
247        self
248    }
249
250    /// Check before starting each file.
251    pub fn is_cancelled(&self) -> bool {
252        self.cancel.is_cancelled()
253    }
254
255    pub fn with_observer(mut self, observer: Arc<dyn OperationObserver>) -> Self {
256        self.observer = Some(observer);
257        self
258    }
259
260    pub(crate) fn observer(&self) -> &dyn OperationObserver {
261        self.observer.as_deref().unwrap_or(&NoopObserver)
262    }
263
264    /// Whether anyone hears the reports, so work done only for them can be skipped.
265    pub(crate) fn is_observed(&self) -> bool {
266        self.observer.is_some()
267    }
268}
269
270pub fn duration_ms(since: Instant) -> u64 {
271    millis(since.elapsed())
272}
273
274pub fn millis(duration: Duration) -> u64 {
275    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
276}
277
278/// Unix milliseconds, for event `ts`.
279pub fn unix_ms(time: SystemTime) -> u64 {
280    time.duration_since(UNIX_EPOCH)
281        .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
282        .unwrap_or(0)
283}
284
285/// RFC 3339 UTC with milliseconds, e.g. `2025-10-21T14:30:00.123Z`.
286pub fn rfc3339_ms(time: SystemTime) -> String {
287    match jiff::Timestamp::try_from(time) {
288        Ok(ts) => format!("{ts:.3}"),
289        Err(_) => format!("{:.3}", jiff::Timestamp::UNIX_EPOCH),
290    }
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    struct Row(RowState);
298    impl OperationRow for Row {
299        fn state(&self) -> RowState {
300            self.0
301        }
302    }
303
304    #[test]
305    fn counts_split_states_and_a_cancel_decides_the_outcome() {
306        use RowState::*;
307        let counts = Counts::of(&[Row(Succeeded), Row(Failed), Row(Cancelled)]);
308        assert_eq!(
309            counts,
310            Counts {
311                total: 3,
312                succeeded: 1,
313                failed: 1,
314                cancelled: 1
315            }
316        );
317        // Cancelled even when every file had already started
318        let started = Counts::of(&[Row(Succeeded)]);
319        assert_eq!(RunOutcome::of(&started, true), RunOutcome::Cancelled);
320        assert_eq!(RunOutcome::of(&started, false), RunOutcome::Complete);
321        let outcome = |rows: &[Row]| RunOutcome::of(&Counts::of(rows), false);
322        assert_eq!(outcome(&[Row(Failed)]), RunOutcome::Failed);
323        assert_eq!(outcome(&[Row(Succeeded), Row(Failed)]), RunOutcome::Partial);
324        assert_eq!(outcome(&[]), RunOutcome::Complete);
325    }
326
327    #[test]
328    fn started_hashing_finishes_on_an_early_error_exit() {
329        let recorder = crate::testutil::Recorder::default();
330        let early_exit = || -> Result<(), ()> {
331            let hashing = Hashing::new(&recorder);
332            hashing.begin();
333            Err(())?;
334            Ok(())
335        };
336        assert!(early_exit().is_err());
337        assert_eq!(recorder.seen(), ["hash start", "hash end"]);
338    }
339}
dvs/operation.rs

dvs/
operation.rs
