mimick/
sync_index.rs

1//! Maintains a persistent index of files that have been previously synced.
2//!
3//! The index maps local file paths to their checksums, modification times,
4//! and target album IDs. A sharded design (`ShardedSyncIndex`) distributes
5//! entries across multiple segment files for lock-free concurrent reads
6//! during parallel scans. On startup, unchanged files are skipped entirely
7//! by comparing stored checksums against the current state.
8
9use serde::{Deserialize, Serialize};
10use std::collections::{HashMap, HashSet};
11use std::fs;
12use std::hash::{DefaultHasher, Hash, Hasher};
13use std::io;
14use std::path::{Path, PathBuf};
15use std::sync::RwLock;
16use std::time::UNIX_EPOCH;
17
18/// Number of shards for distributing sync index entries.
19const SHARD_COUNT: usize = 16;
20
21/// Represents a record stored on disk for a synced file, including the last associated album target.
22#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
23pub struct SyncedFileRecord {
24    /// File size in bytes.
25    pub size: u64,
26    /// Last modified epoch timestamp in milliseconds.
27    pub modified_ms: u64,
28    /// Local pre-computed hex-encoded file SHA-1 checksum.
29    pub checksum: String,
30    /// Targeted Immich album name.
31    #[serde(default)]
32    pub album_name: Option<String>,
33    /// Mapped Immich album unique identifier.
34    #[serde(default)]
35    pub album_id: Option<String>,
36}
37
38/// Describes the intended album target for a file during sync operations.
39#[derive(Clone, Debug, PartialEq, Eq)]
40pub struct SyncTarget {
41    /// Target Immich album name.
42    pub album_name: Option<String>,
43    /// Mapped Immich album unique identifier.
44    pub album_id: Option<String>,
45}
46
47/// Indicates the result of comparing a file on disk with the saved sync index.
48pub enum SyncDecision {
49    UpToDate,
50    NeedsUpload,
51    NeedsReassociate,
52}
53
54/// Inner serialized structure representing the index files list.
55#[derive(Serialize, Deserialize, Default)]
56struct SyncIndexData {
57    /// Mapping of paths to synced file records.
58    files: HashMap<String, SyncedFileRecord>,
59}
60
61/// Borrowed reference of the serialized index for zero-copy writes.
62#[derive(Serialize)]
63struct SyncIndexDataRef<'a> {
64    /// Borrowed mapping of paths to synced file records.
65    files: &'a HashMap<String, SyncedFileRecord>,
66}
67
68/// A single shard of the sync index, holding a subset of entries.
69struct Shard {
70    /// Synced records stored in this shard.
71    entries: HashMap<String, SyncedFileRecord>,
72    /// Reverse index map from checksum to absolute path.
73    checksum_to_path: HashMap<String, String>,
74    /// True if this shard has modified records not yet flushed.
75    dirty: bool,
76}
77
78impl Shard {
79    /// Initialize a new, empty Shard.
80    fn new() -> Self {
81        Self {
82            entries: HashMap::new(),
83            checksum_to_path: HashMap::new(),
84            dirty: false,
85        }
86    }
87}
88
89/// Thread-safe sync index that distributes entries across 16 `RwLock` shards
90/// keyed by path hash. This eliminates the single-lock bottleneck when
91/// parallel producers (startup scan, watcher workers) and consumers (library
92/// UI sync-state lookups) all contend on the index simultaneously.
93///
94/// The public API is identical to the old `SyncIndex` but methods are inherent
95/// -- callers no longer wrap this in `Arc<Mutex<_>>`.
96pub struct ShardedSyncIndex {
97    /// Absolute path of index JSON on disk.
98    index_file: PathBuf,
99    /// Array of RwLock-wrapped entry shards.
100    shards: [RwLock<Shard>; SHARD_COUNT],
101}
102
103/// Determine which shard index a path belongs to.
104fn shard_for(path: &str) -> usize {
105    let mut hasher = DefaultHasher::new();
106    path.hash(&mut hasher);
107    (hasher.finish() as usize) & (SHARD_COUNT - 1)
108}
109
110impl ShardedSyncIndex {
111    /// Load existing indices from persistent path, migrating cache if necessary.
112    pub fn new() -> Self {
113        let index_file = crate::profile::data_dir()
114            .unwrap_or_else(|| PathBuf::from("/tmp").join(crate::profile::dir_segment()))
115            .join("synced_index.json");
116
117        migrate_from_cache_dir(&index_file);
118
119        let all_entries = load_entries(&index_file);
120
121        // Distribute entries across shards.
122        let shards: [RwLock<Shard>; SHARD_COUNT] =
123            std::array::from_fn(|_| RwLock::new(Shard::new()));
124        for (path, record) in all_entries {
125            let idx = shard_for(&path);
126            let mut s = shards[idx].write().unwrap();
127            s.checksum_to_path
128                .insert(record.checksum.clone(), path.clone());
129            s.entries.insert(path, record);
130        }
131
132        Self { index_file, shards }
133    }
134
135    /// Decide whether a file is already current, needs a new upload, or only needs reassociation.
136    pub fn sync_decision(&self, path: &Path, target: &SyncTarget) -> io::Result<SyncDecision> {
137        let metadata = fs::metadata(path)?;
138        let fingerprint = fingerprint_from_metadata(&metadata);
139        let key = path.to_string_lossy();
140        let idx = shard_for(&key);
141        let shard = self.shards[idx].read().unwrap();
142
143        Ok(match shard.entries.get(key.as_ref()) {
144            Some(record) => {
145                if record.size != fingerprint.0 || record.modified_ms != fingerprint.1 {
146                    SyncDecision::NeedsUpload
147                } else if record.album_name != target.album_name
148                    || record.album_id != target.album_id
149                {
150                    SyncDecision::NeedsReassociate
151                } else {
152                    SyncDecision::UpToDate
153                }
154            }
155            None => SyncDecision::NeedsUpload,
156        })
157    }
158
159    /// Save the latest synced fingerprint and album target for a file.
160    pub fn record_synced(&self, path: &str, checksum: &str, target: &SyncTarget) -> io::Result<()> {
161        let metadata = fs::metadata(path)?;
162        let (size, modified_ms) = fingerprint_from_metadata(&metadata);
163        let idx = shard_for(path);
164        let mut shard = self.shards[idx].write().unwrap();
165        shard.entries.insert(
166            path.to_string(),
167            SyncedFileRecord {
168                size,
169                modified_ms,
170                checksum: checksum.to_string(),
171                album_name: target.album_name.clone(),
172                album_id: target.album_id.clone(),
173            },
174        );
175        shard
176            .checksum_to_path
177            .insert(checksum.to_string(), path.to_string());
178        shard.dirty = true;
179        Ok(())
180    }
181
182    /// Drop records for files that no longer exist under any configured watch path.
183    pub fn prune_missing(&self, seen_paths: &HashSet<String>) -> io::Result<()> {
184        let mut any_changed = false;
185        for lock in &self.shards {
186            let mut shard = lock.write().unwrap();
187            let before = shard.entries.len();
188            shard.entries.retain(|path, _| seen_paths.contains(path));
189            shard
190                .checksum_to_path
191                .retain(|_, path| seen_paths.contains(path));
192            if shard.entries.len() != before {
193                shard.dirty = true;
194                any_changed = true;
195            }
196        }
197        if any_changed {
198            self.flush()?;
199        }
200        Ok(())
201    }
202
203    /// Reuse the previous checksum when a file only needs album reassociation.
204    pub fn stored_checksum(&self, path: &str) -> Option<String> {
205        let idx = shard_for(path);
206        let shard = self.shards[idx].read().unwrap();
207        shard
208            .entries
209            .get(path)
210            .map(|record| record.checksum.clone())
211    }
212
213    /// Return the cached checksum only if the file's current size + mtime still
214    /// match the stored fingerprint. Returns `None` when the file is missing,
215    /// untracked, or has been modified since the last index update — callers
216    /// must re-hash in that case.
217    pub fn fresh_checksum(&self, path: &Path) -> Option<String> {
218        let metadata = fs::metadata(path).ok()?;
219        let fingerprint = fingerprint_from_metadata(&metadata);
220        let key = path.to_string_lossy();
221        let idx = shard_for(&key);
222        let shard = self.shards[idx].read().unwrap();
223        let record = shard.entries.get(key.as_ref())?;
224        if record.size == fingerprint.0 && record.modified_ms == fingerprint.1 {
225            Some(record.checksum.clone())
226        } else {
227            None
228        }
229    }
230
231    /// Return a synced record for a specific path without touching the filesystem.
232    pub fn record_for_path(&self, path: &str) -> Option<SyncedFileRecord> {
233        let idx = shard_for(path);
234        let shard = self.shards[idx].read().unwrap();
235        shard.entries.get(path).cloned()
236    }
237
238    /// Remove one synced path from the index after a deliberate delete sync.
239    pub fn remove_path(&self, path: &str) -> io::Result<bool> {
240        let idx = shard_for(path);
241        let mut shard = self.shards[idx].write().unwrap();
242        let Some(record) = shard.entries.remove(path) else {
243            return Ok(false);
244        };
245        let should_remove_checksum = shard
246            .checksum_to_path
247            .get(&record.checksum)
248            .is_some_and(|record_path| record_path == path);
249        if should_remove_checksum {
250            shard.checksum_to_path.remove(&record.checksum);
251        }
252        shard.dirty = true;
253        drop(shard);
254        self.flush()?;
255        Ok(true)
256    }
257
258    /// Return synced records rooted under a watch path.
259    pub fn records_under_path(&self, root: &Path) -> Vec<(String, SyncedFileRecord)> {
260        let mut records = Vec::new();
261        for lock in &self.shards {
262            let shard = lock.read().unwrap();
263            records.extend(
264                shard
265                    .entries
266                    .iter()
267                    .filter(|(path, _)| Path::new(path.as_str()).starts_with(root))
268                    .map(|(path, record)| (path.clone(), record.clone())),
269            );
270        }
271        records
272    }
273
274    /// All synced paths that share a checksum. Used to detect when an asset
275    /// is referenced from more than one watch folder before trashing it.
276    pub fn paths_for_checksum(&self, checksum: &str) -> Vec<String> {
277        let mut paths = Vec::new();
278        for lock in &self.shards {
279            let shard = lock.read().unwrap();
280            for (path, record) in &shard.entries {
281                if record.checksum == checksum {
282                    paths.push(path.clone());
283                }
284            }
285        }
286        paths
287    }
288
289    /// Reverse-lookup a local path by checksum for library sync-state indicators.
290    pub fn local_path_for_checksum(&self, checksum: &str) -> Option<String> {
291        // The checksum could be in any shard, so we must search all.
292        for lock in &self.shards {
293            let shard = lock.read().unwrap();
294            if let Some(path) = shard.checksum_to_path.get(checksum) {
295                return Some(path.clone());
296            }
297        }
298        None
299    }
300
301    /// Force a flush to disk if there are unwritten changes.
302    /// Dirty markers are only cleared after a successful write, so a failed
303    /// write leaves the data marked dirty for the next flush attempt.
304    pub fn flush(&self) -> io::Result<()> {
305        let mut merged = HashMap::new();
306        let mut any_dirty = false;
307        for lock in &self.shards {
308            let shard = lock.read().unwrap();
309            if shard.dirty {
310                any_dirty = true;
311            }
312            merged.extend(shard.entries.iter().map(|(k, v)| (k.clone(), v.clone())));
313        }
314        if !any_dirty {
315            return Ok(());
316        }
317        let content = serde_json::to_string_pretty(&SyncIndexDataRef { files: &merged })?;
318        crate::util::atomic_write(&self.index_file, content.as_bytes())?;
319        for lock in &self.shards {
320            let mut shard = lock.write().unwrap();
321            shard.dirty = false;
322        }
323        Ok(())
324    }
325}
326
327/// Migrate legacy cache-dir synced indices on first launch.
328fn migrate_from_cache_dir(new_path: &Path) {
329    if new_path.exists() {
330        return;
331    }
332    let Some(old_path) = crate::profile::cache_dir().map(|d| d.join("synced_index.json")) else {
333        return;
334    };
335    if !old_path.exists() {
336        return;
337    }
338    if let Some(parent) = new_path.parent()
339        && let Err(err) = fs::create_dir_all(parent)
340    {
341        log::warn!(
342            "Could not create sync index data dir '{}': {}",
343            parent.display(),
344            err
345        );
346        return;
347    }
348    match fs::rename(&old_path, new_path) {
349        Ok(()) => log::info!(
350            "Migrated sync index '{}' -> '{}'",
351            old_path.display(),
352            new_path.display()
353        ),
354        Err(rename_err) => match fs::copy(&old_path, new_path) {
355            Ok(_) => {
356                let _ = fs::remove_file(&old_path);
357                log::info!(
358                    "Copied sync index '{}' -> '{}' (rename failed: {})",
359                    old_path.display(),
360                    new_path.display(),
361                    rename_err
362                );
363            }
364            Err(copy_err) => log::warn!(
365                "Sync index migration failed (rename: {}, copy: {})",
366                rename_err,
367                copy_err
368            ),
369        },
370    }
371}
372
373/// Helper to load all stored sync records from the file.
374fn load_entries(index_file: &Path) -> HashMap<String, SyncedFileRecord> {
375    match fs::read_to_string(index_file) {
376        Ok(content) => match serde_json::from_str::<SyncIndexData>(&content) {
377            Ok(data) => data.files,
378            Err(err) => {
379                log::warn!(
380                    "Failed to parse sync index '{}': {}",
381                    index_file.display(),
382                    err
383                );
384                HashMap::new()
385            }
386        },
387        Err(_) => HashMap::new(),
388    }
389}
390
391/// Helper to calculate the size and modification timestamp fingerprint of a file.
392fn fingerprint_from_metadata(metadata: &fs::Metadata) -> (u64, u64) {
393    let modified_ms = metadata
394        .modified()
395        .ok()
396        .and_then(|mtime| mtime.duration_since(UNIX_EPOCH).ok())
397        .map(|duration| duration.as_millis() as u64)
398        .unwrap_or_default();
399
400    (metadata.len(), modified_ms)
401}
402
403#[cfg(test)]
404mod tests {
405    use super::{Shard, ShardedSyncIndex, SyncDecision, SyncTarget, load_entries, shard_for};
406    use std::collections::HashSet;
407    use std::fs;
408    use std::io::Write;
409    use tempfile::tempdir;
410
411    fn make_index(dir: &std::path::Path) -> ShardedSyncIndex {
412        ShardedSyncIndex {
413            index_file: dir.join("synced_index.json"),
414            shards: std::array::from_fn(|_| std::sync::RwLock::new(Shard::new())),
415        }
416    }
417
418    #[test]
419    fn test_record_synced_then_skip_unchanged_file() {
420        let dir = tempdir().unwrap();
421        let file_path = dir.path().join("photo.jpg");
422        fs::write(&file_path, b"hello").unwrap();
423
424        let index = make_index(dir.path());
425        let target = SyncTarget {
426            album_name: Some("Album".into()),
427            album_id: Some("album-1".into()),
428        };
429
430        assert!(matches!(
431            index.sync_decision(&file_path, &target).unwrap(),
432            SyncDecision::NeedsUpload
433        ));
434        index
435            .record_synced(file_path.to_str().unwrap(), "hash1", &target)
436            .unwrap();
437        assert!(matches!(
438            index.sync_decision(&file_path, &target).unwrap(),
439            SyncDecision::UpToDate
440        ));
441    }
442
443    #[test]
444    fn test_modified_file_needs_resync() {
445        let dir = tempdir().unwrap();
446        let file_path = dir.path().join("photo.jpg");
447        fs::write(&file_path, b"hello").unwrap();
448
449        let index = make_index(dir.path());
450        let target = SyncTarget {
451            album_name: Some("Album".into()),
452            album_id: Some("album-1".into()),
453        };
454        index
455            .record_synced(file_path.to_str().unwrap(), "hash1", &target)
456            .unwrap();
457
458        let mut file = fs::OpenOptions::new()
459            .append(true)
460            .open(&file_path)
461            .unwrap();
462        file.write_all(b" world").unwrap();
463
464        assert!(matches!(
465            index.sync_decision(&file_path, &target).unwrap(),
466            SyncDecision::NeedsUpload
467        ));
468    }
469
470    #[test]
471    fn test_prune_missing_removes_deleted_entries() {
472        let dir = tempdir().unwrap();
473        let index = make_index(dir.path());
474
475        let file_path = dir.path().join("photo.jpg");
476        fs::write(&file_path, b"hello").unwrap();
477        let target = SyncTarget {
478            album_name: Some("Album".into()),
479            album_id: Some("album-1".into()),
480        };
481        index
482            .record_synced(file_path.to_str().unwrap(), "hash1", &target)
483            .unwrap();
484
485        index.prune_missing(&HashSet::new()).unwrap();
486        assert!(matches!(
487            index.sync_decision(&file_path, &target).unwrap(),
488            SyncDecision::NeedsUpload
489        ));
490    }
491
492    #[test]
493    fn test_album_change_requires_reassociate() {
494        let dir = tempdir().unwrap();
495        let file_path = dir.path().join("photo.jpg");
496        fs::write(&file_path, b"hello").unwrap();
497
498        let index = make_index(dir.path());
499        let original = SyncTarget {
500            album_name: Some("Album A".into()),
501            album_id: Some("album-a".into()),
502        };
503        let updated = SyncTarget {
504            album_name: Some("Album B".into()),
505            album_id: Some("album-b".into()),
506        };
507
508        index
509            .record_synced(file_path.to_str().unwrap(), "hash1", &original)
510            .unwrap();
511
512        assert!(matches!(
513            index.sync_decision(&file_path, &updated).unwrap(),
514            SyncDecision::NeedsReassociate
515        ));
516    }
517
518    #[test]
519    fn test_local_path_for_checksum_returns_matching_entry() {
520        let dir = tempdir().unwrap();
521        let index = make_index(dir.path());
522        let file_path = dir.path().join("photo.jpg");
523        fs::write(&file_path, b"hello").unwrap();
524        let target = SyncTarget {
525            album_name: None,
526            album_id: None,
527        };
528        index
529            .record_synced(file_path.to_str().unwrap(), "hash1", &target)
530            .unwrap();
531
532        assert_eq!(
533            index.local_path_for_checksum("hash1"),
534            Some(file_path.to_string_lossy().to_string())
535        );
536        assert!(index.local_path_for_checksum("missing").is_none());
537    }
538
539    #[test]
540    fn test_sharded_index_distributes_entries_across_shards() {
541        let paths = [
542            "/home/user/photos/a.jpg",
543            "/home/user/photos/b.jpg",
544            "/home/user/photos/c.jpg",
545            "/home/user/photos/d.jpg",
546            "/home/user/photos/e.jpg",
547            "/home/user/photos/f.jpg",
548            "/home/user/photos/g.jpg",
549            "/home/user/photos/h.jpg",
550        ];
551        let mut shard_ids: HashSet<usize> = HashSet::new();
552        for path in &paths {
553            shard_ids.insert(shard_for(path));
554        }
555        // With 8 paths and 16 shards, we should have at least 2 distinct shards.
556        assert!(
557            shard_ids.len() >= 2,
558            "Expected at least 2 distinct shards, got {}",
559            shard_ids.len()
560        );
561    }
562
563    #[test]
564    fn test_sharded_index_flush_round_trips() {
565        let dir = tempdir().unwrap();
566        let index = make_index(dir.path());
567
568        let file_path = dir.path().join("photo.jpg");
569        fs::write(&file_path, b"hello").unwrap();
570        let target = SyncTarget {
571            album_name: Some("Album".into()),
572            album_id: Some("album-1".into()),
573        };
574        index
575            .record_synced(file_path.to_str().unwrap(), "hash1", &target)
576            .unwrap();
577        index.flush().unwrap();
578
579        // Load a fresh index from the same file and verify the entry survived.
580        let index_file = dir.path().join("synced_index.json");
581        let all = load_entries(&index_file);
582        let shards = std::array::from_fn(|_| std::sync::RwLock::new(Shard::new()));
583        for (path, record) in all {
584            let idx = shard_for(&path);
585            let mut s = shards[idx].write().unwrap();
586            s.checksum_to_path
587                .insert(record.checksum.clone(), path.clone());
588            s.entries.insert(path, record);
589        }
590        let index2 = ShardedSyncIndex { index_file, shards };
591        assert!(matches!(
592            index2.sync_decision(&file_path, &target).unwrap(),
593            SyncDecision::UpToDate
594        ));
595    }
596}
mimick/sync_index.rs

mimick/
sync_index.rs
