fxcp_core/
merkle_index.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025-2026 Joel Wiramu Pauling <aenertia@aenertia.net>
3//
4// fxcp-core/src/merkle_index.rs  --  Per-target Merkle signature index
5
6//! Stores large Merkle signatures that exceed the 64KB xattr limit in a
7//! consolidated per-target index file at `<target>/.foxing_meta/merkle.db`.
8//!
9//! One file per target, keyed by relative path. Avoids per-file sidecar
10//! proliferation and inode waste for large files.
11
12use std::collections::HashMap;
13use std::io;
14use std::path::{Path, PathBuf};
15use std::sync::{Arc, LazyLock};
16use dashmap::DashMap;
17use tracing::{debug, warn};
18
19use crate::hashing::MerkleSignature;
20
21const META_DIR: &str = ".foxing_meta";
22const INDEX_FILE: &str = "merkle.db";
23
24/// Global in-memory cache of per-target Merkle indexes.
25/// Key: target root path, Value: map of rel_path -> MerkleSignature.
26static INDEX_CACHE: LazyLock<DashMap<PathBuf, Arc<DashMap<PathBuf, MerkleSignature>>>> =
27    LazyLock::new(DashMap::new);
28
29/// Chunk size for large-file Merkle signatures (re-exported from constants).
30pub use crate::constants::MERKLE_INDEX_CHUNK_SIZE as LARGE_FILE_CHUNK_SIZE;
31/// File size threshold for using the per-target Merkle index (re-exported from constants).
32pub use crate::constants::MERKLE_INDEX_THRESHOLD as INDEX_THRESHOLD;
33
34/// Store a Merkle signature in the per-target index (auto-flushes to disk).
35pub fn set(target_root: &Path, rel_path: &Path, sig: &MerkleSignature) -> io::Result<()> {
36    let index = get_or_load_index(target_root);
37    index.insert(rel_path.to_path_buf(), sig.clone());
38    flush_index(target_root, &index)
39}
40
41/// Insert into the in-memory index WITHOUT flushing to disk.
42/// Caller must call [`flush()`] or [`flush_all()`] explicitly to persist.
43pub fn batch_set(target_root: &Path, rel_path: &Path, sig: &MerkleSignature) -> io::Result<()> {
44    let index = get_or_load_index(target_root);
45    index.insert(rel_path.to_path_buf(), sig.clone());
46    Ok(())
47}
48
49/// Retrieve a Merkle signature from the per-target index.
50pub fn get(target_root: &Path, rel_path: &Path) -> Option<MerkleSignature> {
51    let index = get_or_load_index(target_root);
52    index.get(rel_path).map(|v| v.clone())
53}
54
55/// Remove a Merkle signature from the index (file deleted or moved).
56pub fn remove(target_root: &Path, rel_path: &Path) -> io::Result<()> {
57    let index = get_or_load_index(target_root);
58    index.remove(rel_path);
59    flush_index(target_root, &index)
60}
61
62/// Explicitly flush a single target's index to disk (crash-safe).
63pub fn flush(target_root: &Path) -> io::Result<()> {
64    let index = get_or_load_index(target_root);
65    flush_index(target_root, &index)
66}
67
68/// Flush ALL targets in the global index cache to disk.
69pub fn flush_all() -> io::Result<()> {
70    let mut last_err = None;
71    for entry in INDEX_CACHE.iter() {
72        if let Err(e) = flush_index(entry.key(), entry.value()) {
73            last_err = Some(e);
74        }
75    }
76    last_err.map_or(Ok(()), Err)
77}
78
79/// RAII guard that calls [`flush_all()`] on drop for daemon shutdown safety.
80pub struct FlushGuard;
81
82impl Drop for FlushGuard {
83    fn drop(&mut self) {
84        let _ = flush_all();
85    }
86}
87
88/// Create a [`FlushGuard`] that flushes all indexes on drop.
89pub fn flush_guard() -> FlushGuard {
90    FlushGuard
91}
92
93fn index_path(target_root: &Path) -> PathBuf {
94    target_root.join(META_DIR).join(INDEX_FILE)
95}
96
97fn get_or_load_index(target_root: &Path) -> Arc<DashMap<PathBuf, MerkleSignature>> {
98    INDEX_CACHE.entry(target_root.to_path_buf())
99        .or_insert_with(|| Arc::new(load_index(target_root)))
100        .clone()
101}
102
103fn load_index(target_root: &Path) -> DashMap<PathBuf, MerkleSignature> {
104    let path = index_path(target_root);
105    let map = DashMap::new();
106
107    if let Ok(data) = std::fs::read(&path) {
108        match bincode::deserialize::<HashMap<PathBuf, MerkleSignature>>(&data) {
109            Ok(stored) => {
110                debug!("merkle_index: Loaded {} entries from {:?}", stored.len(), path);
111                for (k, v) in stored {
112                    map.insert(k, v);
113                }
114            }
115            Err(e) => {
116                warn!("merkle_index: Failed to deserialize {:?}: {}. Starting fresh.", path, e);
117            }
118        }
119    }
120
121    map
122}
123
124fn flush_index(target_root: &Path, index: &DashMap<PathBuf, MerkleSignature>) -> io::Result<()> {
125    let meta_dir = target_root.join(META_DIR);
126    std::fs::create_dir_all(&meta_dir)?;
127
128    // Collect into a HashMap for serialization
129    let snapshot: HashMap<PathBuf, MerkleSignature> = index.iter()
130        .map(|entry| (entry.key().clone(), entry.value().clone()))
131        .collect();
132
133    let data = bincode::serialize(&snapshot)
134        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
135
136    // Crash-safe atomic write: tmp + fsync + rename + dir fsync
137    // Per-call unique tmp prevents concurrent flush_all/flush_index race on same file
138    static FLUSH_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
139    let path = index_path(target_root);
140    let seq = FLUSH_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
141    let tmp = path.with_extension(format!("db.tmp.{}", seq));
142    std::fs::write(&tmp, &data)?;
143    {
144        let file = std::fs::File::open(&tmp)?;
145        file.sync_data()?;
146    }
147    std::fs::rename(&tmp, &path)?;
148    if let Some(parent) = path.parent()
149        && let Ok(dir) = std::fs::File::open(parent) {
150            let _ = dir.sync_data();
151        }
152
153    debug!("merkle_index: Flushed {} entries ({} bytes) to {:?}",
154           snapshot.len(), data.len(), path);
155    Ok(())
156}
157
158/// Invalidate the in-memory cache for a target (e.g., on target reconnect).
159pub fn invalidate_cache(target_root: &Path) {
160    INDEX_CACHE.remove(target_root);
161}
162
163#[cfg(test)]
164pub fn clear_cache() {
165    INDEX_CACHE.clear();
166}
167
168#[cfg(test)]
169mod tests {
170    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
171    use super::*;
172    use tempfile::TempDir;
173
174    fn make_sig(root: [u8; 32]) -> MerkleSignature {
175        MerkleSignature {
176            root,
177            chunk_size: 65536,
178            file_size: 1024 * 1024,
179            leaf_hashes: vec![root],
180            chunk_entropies: None,
181            chunk_simhashes: None,
182        }
183    }
184
185    #[test]
186    fn test_set_and_get_roundtrip() {
187        let dir = TempDir::new().unwrap();
188        let target = dir.path();
189
190        let rel = Path::new("data/bigfile.bin");
191        let sig = make_sig([0xAA; 32]);
192        set(target, rel, &sig).unwrap();
193
194        let got = get(target, rel).expect("should return Some");
195        assert_eq!(got.root, sig.root);
196        assert_eq!(got.chunk_size, sig.chunk_size);
197        assert_eq!(got.file_size, sig.file_size);
198        assert_eq!(got.leaf_hashes, sig.leaf_hashes);
199
200        super::invalidate_cache(target);
201    }
202
203    #[test]
204    fn test_set_overwrites_existing() {
205        let dir = TempDir::new().unwrap();
206        let target = dir.path();
207
208        let rel = Path::new("overwrite.bin");
209        let sig1 = make_sig([0x11; 32]);
210        let sig2 = make_sig([0x22; 32]);
211
212        set(target, rel, &sig1).unwrap();
213        set(target, rel, &sig2).unwrap();
214
215        let got = get(target, rel).expect("should return sig2");
216        assert_eq!(got.root, [0x22; 32]);
217
218        super::invalidate_cache(target);
219    }
220
221    #[test]
222    fn test_get_nonexistent_returns_none() {
223        let dir = TempDir::new().unwrap();
224        let target = dir.path();
225
226        assert!(get(target, Path::new("no/such/file.bin")).is_none());
227
228        super::invalidate_cache(target);
229    }
230
231    #[test]
232    fn test_remove_makes_get_none() {
233        let dir = TempDir::new().unwrap();
234        let target = dir.path();
235
236        let rel = Path::new("removeme.bin");
237        let sig = make_sig([0xBB; 32]);
238        set(target, rel, &sig).unwrap();
239        assert!(get(target, rel).is_some());
240
241        remove(target, rel).unwrap();
242        assert!(get(target, rel).is_none());
243
244        super::invalidate_cache(target);
245    }
246
247    #[test]
248    fn test_flush_creates_db_file() {
249        let dir = TempDir::new().unwrap();
250        let target = dir.path();
251
252        let rel = Path::new("flushed.bin");
253        let sig = make_sig([0xCC; 32]);
254        set(target, rel, &sig).unwrap();
255
256        let db_path = target.join(META_DIR).join(INDEX_FILE);
257        assert!(db_path.exists(), "merkle.db should exist after set()");
258        assert!(std::fs::metadata(&db_path).unwrap().len() > 0);
259
260        super::invalidate_cache(target);
261    }
262
263    #[test]
264    fn test_load_from_disk_after_invalidate() {
265        let dir = TempDir::new().unwrap();
266        let target = dir.path();
267
268        let rel = Path::new("persist.bin");
269        let sig = make_sig([0xDD; 32]);
270        set(target, rel, &sig).unwrap();
271
272        // Evict from in-memory cache
273        invalidate_cache(target);
274
275        // get() should reload from disk
276        let got = get(target, rel).expect("should reload from disk");
277        assert_eq!(got.root, sig.root);
278        assert_eq!(got.chunk_size, sig.chunk_size);
279        assert_eq!(got.file_size, sig.file_size);
280        assert_eq!(got.leaf_hashes, sig.leaf_hashes);
281
282        super::invalidate_cache(target);
283    }
284
285    #[test]
286    fn test_empty_index_serializes_deserializes() {
287        let dir = TempDir::new().unwrap();
288        let target = dir.path();
289
290        let rel = Path::new("temp.bin");
291        let sig = make_sig([0xEE; 32]);
292        // Insert then remove to trigger flush of an empty index
293        set(target, rel, &sig).unwrap();
294        remove(target, rel).unwrap();
295
296        let db_path = target.join(META_DIR).join(INDEX_FILE);
297        assert!(db_path.exists(), "merkle.db should exist even when empty");
298
299        // Verify the on-disk data deserializes to an empty HashMap
300        let data = std::fs::read(&db_path).unwrap();
301        let stored: HashMap<PathBuf, MerkleSignature> =
302            bincode::deserialize(&data).expect("should deserialize");
303        assert!(stored.is_empty(), "index should be empty after remove");
304
305        super::invalidate_cache(target);
306    }
307
308    #[test]
309    fn test_concurrent_set_all_present() {
310        let dir = TempDir::new().unwrap();
311        let target = dir.path();
312
313        let count = 20usize;
314        for i in 0..count {
315            let rel = PathBuf::from(format!("file_{:04}.bin", i));
316            let mut root = [0u8; 32];
317            root[0] = i as u8;
318            let sig = make_sig(root);
319            set(target, &rel, &sig).unwrap();
320        }
321
322        // Verify all 20 are present via get()
323        for i in 0..count {
324            let rel = PathBuf::from(format!("file_{:04}.bin", i));
325            let got = get(target, &rel)
326                .unwrap_or_else(|| panic!("file_{:04}.bin should be present", i));
327            assert_eq!(got.root[0], i as u8);
328        }
329
330        // Also verify the on-disk state
331        invalidate_cache(target);
332        for i in 0..count {
333            let rel = PathBuf::from(format!("file_{:04}.bin", i));
334            let got = get(target, &rel)
335                .unwrap_or_else(|| panic!("file_{:04}.bin should survive reload", i));
336            assert_eq!(got.root[0], i as u8);
337        }
338
339        super::invalidate_cache(target);
340    }
341
342    #[test]
343    fn test_invalidate_then_reload() {
344        let dir = TempDir::new().unwrap();
345        let target = dir.path();
346
347        let rel = Path::new("reload.bin");
348        let sig = make_sig([0xFF; 32]);
349        set(target, rel, &sig).unwrap();
350
351        // Flush is implicit in set(). Now invalidate and reload.
352        invalidate_cache(target);
353
354        let got = get(target, rel).expect("should reload after invalidate");
355        assert_eq!(got.root, [0xFF; 32]);
356        assert_eq!(got.file_size, 1024 * 1024);
357        assert_eq!(got.chunk_size, 65536);
358        assert_eq!(got.leaf_hashes.len(), 1);
359
360        super::invalidate_cache(target);
361    }
362
363    #[test]
364    fn test_set_multiple_entries_persist() {
365        let dir = TempDir::new().unwrap();
366        let target = dir.path();
367
368        for i in 0..5u8 {
369            let rel = PathBuf::from(format!("multi_{}.bin", i));
370            let mut root = [0u8; 32];
371            root[0] = i;
372            set(target, &rel, &make_sig(root)).unwrap();
373        }
374
375        invalidate_cache(target);
376
377        for i in 0..5u8 {
378            let rel = PathBuf::from(format!("multi_{}.bin", i));
379            let got = get(target, &rel).expect("set entry should persist after invalidate");
380            assert_eq!(got.root[0], i);
381        }
382
383        super::invalidate_cache(target);
384    }
385
386    #[test]
387    fn test_set_multi_target_persist() {
388        let dir1 = TempDir::new().unwrap();
389        let dir2 = TempDir::new().unwrap();
390        let target1 = dir1.path();
391        let target2 = dir2.path();
392
393        set(target1, Path::new("a.bin"), &make_sig([0x11; 32])).unwrap();
394        set(target2, Path::new("b.bin"), &make_sig([0x22; 32])).unwrap();
395
396        invalidate_cache(target1);
397        invalidate_cache(target2);
398
399        let got1 = get(target1, Path::new("a.bin")).expect("target1 entry should persist");
400        assert_eq!(got1.root, [0x11; 32]);
401
402        let got2 = get(target2, Path::new("b.bin")).expect("target2 entry should persist");
403        assert_eq!(got2.root, [0x22; 32]);
404
405        super::invalidate_cache(target1);
406        super::invalidate_cache(target2);
407    }
408
409    #[test]
410    fn test_flush_guard_persists_on_drop() {
411        let dir = TempDir::new().unwrap();
412        let target = dir.path();
413
414        {
415            let _guard = flush_guard();
416            set(target, Path::new("guarded.bin"), &make_sig([0xAB; 32])).unwrap();
417        }
418
419        invalidate_cache(target);
420        let got = get(target, Path::new("guarded.bin")).expect("entry should survive FlushGuard drop");
421        assert_eq!(got.root, [0xAB; 32]);
422
423        super::invalidate_cache(target);
424    }
425}
fxcp_core/merkle_index.rs

fxcp_core/
merkle_index.rs
