diff --git a/crates/engine/src/sync.rs b/crates/engine/src/sync.rs index b07566cafe..3c94e35a83 100644 --- a/crates/engine/src/sync.rs +++ b/crates/engine/src/sync.rs @@ -433,73 +433,226 @@ fn sidecar_has_news(labels: &Catalog, p: &Photo, naming: SidecarNaming) -> bool /// whose file is back by now is left alone. Once the import is in, a part that fails is /// reported in `failed` and the rest is still done (and still one step). pub fn synchronize(s: &mut Session, changes: FolderChanges, choice: SyncChoice) -> Result { - if changes.offline { - return Err(not_there(&changes.path)); + let (work, mut commit) = SyncJob::start(s, changes, choice)?; + let mut steps = Vec::new(); + work.run(&SyncProgress::default(), |step| { + steps.push(step); + true + }); + for step in steps { + commit.apply(s, step); + } + Ok(commit.finish(s)) +} + +/// Synchronizing in halves, so no file is touched on the UI thread: [`SyncJob::start`] takes what +/// it needs from the session (no file-system call); the [`SyncWork`] does every file read (the +/// import's probes, whether files are still where the scan found them, the sidecars) and hands +/// over [`SyncStep`]s — on a worker thread in the app; [`SyncCommit::apply`] puts each into the +/// catalog and [`SyncCommit::finish`] makes them one undo step. +pub struct SyncJob; + +/// The file-system half of a synchronize (see [`SyncJob`]). +pub struct SyncWork { + /// The import of the new files, and the files. + import: Option<(crate::import::ImportJob, Vec)>, + moved: Vec, + missing: Vec, + /// (photo, its file, how its sidecar is named) + sidecars: Vec<(PhotoId, String, SidecarNaming, bool)>, +} + +/// The catalog half of a synchronize (see [`SyncJob`]). +pub struct SyncCommit { + opts: ImportOptions, + now: String, + mark: u64, + report: SyncReport, +} + +/// A piece of work done, for [`SyncCommit::apply`]. +pub enum SyncStep { + /// A batch of new files, readied for the catalog. + Import(crate::import::Prepared), + /// Moved photos whose file is still at its new place (photo, new path). + Relink(Vec<(PhotoId, String)>), + /// Missing photos whose file is still gone. + Remove(Vec), + /// Sidecars read (photo, its file, what the sidecar says or why it couldn't be read). + Sidecars(Vec<(PhotoId, String, std::result::Result)>), +} + +impl SyncJob { + /// Ready a synchronize of `changes` as `choice` says. Reads no file. + pub fn start(s: &mut Session, changes: FolderChanges, choice: SyncChoice) -> Result<(SyncWork, SyncCommit)> { + if changes.offline { + return Err(not_there(&changes.path)); + } + let FolderChanges { new, missing, metadata, moved, probes, .. } = changes; + let opts = ImportOptions { mode: ImportMode::Add, ..Default::default() }; + let import = if choice.import_new && !new.is_empty() { + s.import_probes = probes; + let job = crate::import::ImportJob::new(s, opts.clone())?; + Some((job, new.into_iter().map(|c| c.path).collect())) + } else { + None + }; + let now = import.as_ref().map_or_else(|| (s.clock)(), |(j, _)| j.now().to_string()); + let sidecars = if choice.read_metadata { + metadata + .into_iter() + .filter_map(|m| { + let p = s.catalog.photo(PhotoId(m.id))?; + Some((p.id, m.path, s.sidecar_naming(p.id), p.kind == lightcraft_catalog::MediaKind::Raw)) + }) + .collect() + } else { + Vec::new() + }; + let work = SyncWork { + import, + moved: if choice.relink_moved { moved } else { Vec::new() }, + missing: if choice.remove_missing { missing } else { Vec::new() }, + sidecars, + }; + Ok((work, SyncCommit { opts, now, mark: s.commits(), report: SyncReport::default() })) + } +} + +impl SyncWork { + /// Do the work, handing each step to `send` (`false`: nobody takes them any more — stop). + /// Stops before the next piece once `progress.files.cancel` is set; what was handed over + /// stays handed over. + pub fn run(self, progress: &SyncProgress, mut send: impl FnMut(SyncStep) -> bool) { + use std::sync::atomic::Ordering::Relaxed; + let cancel = &progress.files.cancel; + let SyncWork { import, moved, missing, sidecars } = self; + let pieces = import.as_ref().map_or(0, |(_, f)| f.len()) + moved.len() + missing.len() + sidecars.len(); + progress.to_check.store(pieces, Relaxed); + progress.checked.store(0, Relaxed); + let step = |n: usize| { + progress.checked.fetch_add(n, Relaxed); + }; + if let Some((mut job, files)) = import { + for chunk in files.chunks(crate::import::batch_size()) { + if cancel.load(Relaxed) { + return; + } + let prepared = job.prepare_files(chunk.to_vec(), cancel); + step(chunk.len()); + // (the files are added in place: a batch nobody takes leaves nothing behind on disk) + if !send(SyncStep::Import(prepared)) { + return; + } + } + } + if cancel.load(Relaxed) { + return; + } + // still where the scan found it, and still gone from where the photo says + let relink: Vec<(PhotoId, String)> = + moved.into_iter().filter(|m| Path::new(&m.to).is_file() && !Path::new(&m.from).exists()).map(|m| (PhotoId(m.id), m.to)).collect(); + step(relink.len()); + if !relink.is_empty() && !send(SyncStep::Relink(relink)) { + return; + } + if cancel.load(Relaxed) { + return; + } + let gone: Vec = missing.iter().filter(|m| !Path::new(&m.path).exists()).map(|m| PhotoId(m.id)).collect(); + step(missing.len()); + if !gone.is_empty() && !send(SyncStep::Remove(gone)) { + return; + } + let mut read = Vec::with_capacity(sidecars.len()); + for (id, path, naming, raw) in sidecars { + if cancel.load(Relaxed) { + break; + } + read.push((id, path.clone(), read_sidecar(&path, naming, raw))); + step(1); + } + if !read.is_empty() { + send(SyncStep::Sidecars(read)); + } + } +} + +/// Read and parse photo file `path`'s XMP sidecar file. +fn read_sidecar(path: &str, naming: SidecarNaming, raw: bool) -> std::result::Result { + let file = crate::sidecar::find_sidecar(path, naming).ok_or_else(|| "no XMP sidecar".to_string())?; + let len = std::fs::metadata(&file).map_err(|e| format!("{}: {e}", file.display()))?.len(); + if len > MAX_SIDECAR_BYTES { + return Err(format!("{}: too large for an XMP sidecar", file.display())); + } + let text = std::fs::read_to_string(&file).map_err(|e| format!("{}: {e}", file.display()))?; + crate::sidecar::parse_sidecar(&text, raw).map_err(|e| format!("{}: {e}", file.display())) +} + +impl SyncCommit { + /// Put a step into the catalog (no file is touched). A part that fails is reported in the + /// end, and the rest is still done. + pub fn apply(&mut self, s: &mut Session, step: SyncStep) { + let report = &mut self.report; + match step { + SyncStep::Import(prepared) => match crate::import::commit_prepared(s, &self.opts, &self.now, prepared) { + Ok(r) => { + report.imported += r.imported.len(); + report.failed.extend(r.failed); + } + Err(e) => report.failed.push((String::new(), e.to_string())), + }, + SyncStep::Relink(moved) => { + let ops: Vec = + moved.iter().filter(|(id, _)| s.catalog.photo(*id).is_some()).map(|(id, to)| crate::cmd::missing::relink_op(*id, to)).collect(); + let n = ops.len(); + if n > 0 { + match s.commit("Relink Moved Photos", Op::Batch { ops }) { + Ok(()) => report.relinked += n, + Err(e) => report.failed.push((String::new(), e.to_string())), + } + } + } + SyncStep::Remove(gone) => { + let ops: Vec = gone + .iter() + .filter(|id| s.catalog.photo(**id).is_some_and(|p| p.in_library())) + .map(|id| Op::SetDeleted { id: *id, deleted: true }) + .collect(); + let n = ops.len(); + if n > 0 { + match s.commit("Remove Missing Photos", Op::Batch { ops }) { + Ok(()) => report.removed += n, + Err(e) => report.failed.push((String::new(), e.to_string())), + } + } + } + SyncStep::Sidecars(read) => { + let mut ops = Vec::new(); + for (id, path, sc) in read { + match sc.map_err(EngineError::Other).and_then(|sc| s.sidecar_op(id, sc)) { + Ok(op) => ops.push(op), + Err(e) => report.failed.push((path, e.to_string())), + } + } + let n = ops.len(); + if n > 0 { + match s.commit("Read Metadata from File", Op::Batch { ops }) { + Ok(()) => report.read += n, + Err(e) => report.failed.push((String::new(), e.to_string())), + } + } + } + } + } + + /// What was done, as one undo step ("Synchronize Folder"). + pub fn finish(self, s: &mut Session) -> SyncReport { + let steps = usize::try_from(s.commits().wrapping_sub(self.mark)).unwrap_or(usize::MAX).min(s.undo.len()); + s.merge_undo(steps, "Synchronize Folder"); + if let Some(last) = s.undo.last_mut().filter(|_| steps == 1) { + last.label = "Synchronize Folder".into(); + } + self.report } - let mark = s.commits(); - let mut report = SyncReport::default(); - let FolderChanges { new, missing, metadata, moved, probes, .. } = changes; - if choice.import_new && !new.is_empty() { - s.import_probes = probes; - let files: Vec = new.into_iter().map(|c| c.path).collect(); - let r = crate::import::import_with(s, &files, &ImportOptions { mode: ImportMode::Add, ..Default::default() })?; - report.imported = r.imported.len(); - report.failed.extend(r.failed); - } - if choice.relink_moved { - // still where the scan found it, and still gone from where the photo says - let ops: Vec = moved - .iter() - .filter(|m| Path::new(&m.to).is_file() && !Path::new(&m.from).exists() && s.catalog.photo(PhotoId(m.id)).is_some()) - .map(|m| crate::cmd::missing::relink_op(PhotoId(m.id), &m.to)) - .collect(); - if !ops.is_empty() { - let n = ops.len(); - match s.commit("Relink Moved Photos", Op::Batch { ops }) { - Ok(()) => report.relinked = n, - Err(e) => report.failed.push((String::new(), e.to_string())), - } - } - } - if choice.remove_missing { - let gone: Vec = missing.iter().filter(|m| !Path::new(&m.path).exists()).map(|m| PhotoId(m.id)).collect(); - let ops: Vec = gone - .iter() - .filter(|id| s.catalog.photo(**id).is_some_and(|p| p.in_library())) - .map(|id| Op::SetDeleted { id: *id, deleted: true }) - .collect(); - if !ops.is_empty() { - let n = ops.len(); - match s.commit("Remove Missing Photos", Op::Batch { ops }) { - Ok(()) => report.removed = n, - Err(e) => report.failed.push((String::new(), e.to_string())), - } - } - } - if choice.read_metadata { - let mut ops = Vec::new(); - for m in &metadata { - match s.read_sidecar_op(PhotoId(m.id)) { - Ok(Some((op, _))) => { - ops.push(op); - report.read += 1; - } - Ok(None) => report.failed.push((m.path.clone(), "no XMP sidecar".into())), - Err(e) => report.failed.push((m.path.clone(), e.to_string())), - } - } - if !ops.is_empty() - && let Err(e) = s.commit("Read Metadata from File", Op::Batch { ops }) - { - report.read = 0; - report.failed.push((String::new(), e.to_string())); - } - } - // (what was done stays one step even when a later part failed: the report says which) - let steps = usize::try_from(s.commits().wrapping_sub(mark)).unwrap_or(usize::MAX).min(s.undo.len()); - s.merge_undo(steps, "Synchronize Folder"); - if let Some(last) = s.undo.last_mut().filter(|_| steps == 1) { - last.label = "Synchronize Folder".into(); - } - Ok(report) } diff --git a/crates/engine/src/tests_sync.rs b/crates/engine/src/tests_sync.rs index 724487ff25..dd6ab7ac13 100644 --- a/crates/engine/src/tests_sync.rs +++ b/crates/engine/src/tests_sync.rs @@ -466,3 +466,75 @@ fn face_regions_found_meanwhile_leave_the_scan_current() { let r = s.execute("folder.synchronize", &json!({"path": dir.path("trip"), "scanned": true})).unwrap(); assert_eq!(r["imported"], 1, "{r}"); } + +/// Synchronizing in halves, as the app does it: readied on the UI thread without touching the +/// disk, the work (every file read) on a worker, each step committed back on the UI thread. +fn sync_in_halves(s: &mut Session, path: &str, choice: crate::sync::SyncChoice) -> crate::sync::SyncReport { + let changes = crate::sync::scan(s, path, false).unwrap(); + let (work, mut commit) = crate::sync::SyncJob::start(s, changes, choice).unwrap(); + let progress = crate::sync::SyncProgress::default(); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || work.run(&progress, |step| tx.send(step).is_ok())).join().unwrap(); + for step in rx { + commit.apply(s, step); + } + commit.finish(s) +} + +#[test] +fn the_work_of_synchronizing_runs_apart_from_its_commits() { + let dir = Scratch::new("halves"); + let mut s = library(&dir); + let before = holdings(&s); + write_png(&dir.path("trip/c.png"), 3); + std::fs::remove_file(dir.path("trip/day1/b.png")).unwrap(); + write_sidecar(&dir.path("trip/a.png"), 5, Duration::ZERO); + let choice = crate::sync::SyncChoice { import_new: true, relink_moved: true, remove_missing: true, read_metadata: true }; + let r = sync_in_halves(&mut s, &dir.path("trip"), choice); + assert_eq!((r.imported, r.removed, r.read), (1, 1, 1), "{r:?}"); + assert_eq!(s.catalog.photos().find(|p| p.file_name == "a.png").unwrap().rating, 5); + s.execute("edit.undo", &json!({})).unwrap(); + assert_eq!(holdings(&s), before, "still one undo step"); +} + +#[test] +fn readying_a_synchronize_reads_no_file() { + let dir = Scratch::new("ready"); + let mut s = library(&dir); + write_png(&dir.path("trip/c.png"), 3); + let changes = crate::sync::scan(&mut s, &dir.path("trip"), false).unwrap(); + // the file goes away between readying (UI thread) and the work (worker): the work finds out + let (work, mut commit) = crate::sync::SyncJob::start(&mut s, changes, crate::sync::SyncChoice::default()).unwrap(); + std::fs::remove_file(dir.path("trip/c.png")).unwrap(); + let mut steps = Vec::new(); + work.run(&crate::sync::SyncProgress::default(), |step| { + steps.push(step); + true + }); + for step in steps { + commit.apply(&mut s, step); + } + let r = commit.finish(&mut s); + assert_eq!(r.imported, 0, "{r:?}"); + assert_eq!(r.failed.len(), 1, "{r:?}"); +} + +#[test] +fn a_cancelled_synchronize_does_nothing_more() { + let dir = Scratch::new("cancel-run"); + let mut s = library(&dir); + let steps_before = s.undo.len(); + write_png(&dir.path("trip/c.png"), 3); + let changes = crate::sync::scan(&mut s, &dir.path("trip"), false).unwrap(); + let (work, commit) = crate::sync::SyncJob::start(&mut s, changes, crate::sync::SyncChoice::default()).unwrap(); + let progress = crate::sync::SyncProgress::default(); + progress.files.cancel.store(true, std::sync::atomic::Ordering::Relaxed); + let mut sent = 0; + work.run(&progress, |_| { + sent += 1; + true + }); + assert_eq!(sent, 0); + let r = commit.finish(&mut s); + assert_eq!((r.imported, s.undo.len()), (0, steps_before)); +}