Skip to main content

hashtree_fs/
lib.rs

1//! Filesystem-based content-addressed blob storage.
2//!
3//! Stores blobs in a directory structure similar to git:
4//! `{base_path}/{first 2 chars of hash}/{next 2 chars}/{remaining hash chars}`
5//!
6//! For example, a blob with hash `abcdef123...` would be stored at:
7//! `~/.hashtree/blobs/ab/cd/ef123...`
8
9use async_trait::async_trait;
10use hashtree_core::store::{Store, StoreError, StoreStats};
11use hashtree_core::types::Hash;
12use std::collections::HashMap;
13use std::fs;
14use std::io::{Read, Seek, SeekFrom};
15use std::path::{Path, PathBuf};
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::sync::RwLock;
18use std::time::{SystemTime, UNIX_EPOCH};
19
20/// Filesystem-backed blob store implementing hashtree's Store trait.
21///
22/// Stores blobs in a two-level sharded directory structure using
23/// the first 4 hex characters of the hash as directory prefixes.
24/// Legacy single-level shard paths remain readable.
25/// Supports storage limits with mtime-based FIFO eviction and pinning.
26pub struct FsBlobStore {
27    base_path: PathBuf,
28    max_bytes: AtomicU64,
29    /// Pin counts stored in memory, persisted to pins.json
30    pins: RwLock<HashMap<String, u32>>,
31}
32
33impl FsBlobStore {
34    /// Create a new filesystem blob store at the given path.
35    ///
36    /// Creates the directory if it doesn't exist.
37    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self, StoreError> {
38        let base_path = path.as_ref().to_path_buf();
39        fs::create_dir_all(&base_path)?;
40
41        // Load existing pins from disk
42        let pins = Self::load_pins(&base_path).unwrap_or_default();
43
44        Ok(Self {
45            base_path,
46            max_bytes: AtomicU64::new(0), // 0 = unlimited
47            pins: RwLock::new(pins),
48        })
49    }
50
51    /// Create a new store with a maximum size limit
52    pub fn with_max_bytes<P: AsRef<Path>>(path: P, max_bytes: u64) -> Result<Self, StoreError> {
53        let store = Self::new(path)?;
54        store.max_bytes.store(max_bytes, Ordering::Relaxed);
55        Ok(store)
56    }
57
58    /// Path to pins.json file
59    fn pins_path(&self) -> PathBuf {
60        self.base_path.join("pins.json")
61    }
62
63    /// Load pins from disk
64    fn load_pins(base_path: &Path) -> Option<HashMap<String, u32>> {
65        let pins_path = base_path.join("pins.json");
66        let contents = fs::read_to_string(pins_path).ok()?;
67        serde_json::from_str(&contents).ok()
68    }
69
70    /// Save pins to disk
71    fn save_pins(&self) -> Result<(), StoreError> {
72        let pins = self.pins.read().unwrap();
73        let json = serde_json::to_string(&*pins)
74            .map_err(|e| StoreError::Other(format!("Failed to serialize pins: {}", e)))?;
75        fs::write(self.pins_path(), json)?;
76        Ok(())
77    }
78
79    fn blob_path_from_hex(&self, hash_hex: &str) -> PathBuf {
80        let (prefix, rest) = hash_hex.split_at(2);
81        let (subdir, filename) = rest.split_at(2);
82        self.base_path.join(prefix).join(subdir).join(filename)
83    }
84
85    fn legacy_blob_path(&self, hash: &Hash) -> PathBuf {
86        let hex = hex::encode(hash);
87        let (prefix, rest) = hex.split_at(2);
88        self.base_path.join(prefix).join(rest)
89    }
90
91    /// Get the file path for a given hash.
92    ///
93    /// Format: `{base_path}/{first 2 hex chars}/{next 2 hex chars}/{remaining 60 hex chars}`
94    fn blob_path(&self, hash: &Hash) -> PathBuf {
95        self.blob_path_from_hex(&hex::encode(hash))
96    }
97
98    fn existing_blob_path(&self, hash: &Hash) -> Option<PathBuf> {
99        let primary = self.blob_path(hash);
100        if primary.exists() {
101            return Some(primary);
102        }
103
104        let legacy = self.legacy_blob_path(hash);
105        if legacy.exists() {
106            return Some(legacy);
107        }
108
109        None
110    }
111
112    fn hash_hex_for_blob_path(&self, path: &Path) -> Option<String> {
113        let relative = path.strip_prefix(&self.base_path).ok()?;
114        let mut hex = String::new();
115
116        for component in relative.iter() {
117            let part = component.to_str()?;
118            hex.push_str(part);
119        }
120
121        if hex.len() != 64 || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
122            return None;
123        }
124
125        Some(hex.to_ascii_lowercase())
126    }
127
128    fn collect_blob_metadata_recursive(
129        &self,
130        dir: &Path,
131        blobs: &mut Vec<(PathBuf, String, fs::Metadata)>,
132    ) -> Result<(), StoreError> {
133        let entries = match fs::read_dir(dir) {
134            Ok(e) => e,
135            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
136            Err(e) => return Err(e.into()),
137        };
138
139        for entry in entries {
140            let entry = entry?;
141            let file_type = entry.file_type()?;
142            let path = entry.path();
143
144            if file_type.is_dir() {
145                self.collect_blob_metadata_recursive(&path, blobs)?;
146                continue;
147            }
148
149            if !file_type.is_file() {
150                continue;
151            }
152
153            if let Some(hex) = self.hash_hex_for_blob_path(&path) {
154                blobs.push((path, hex, entry.metadata()?));
155            }
156        }
157
158        Ok(())
159    }
160
161    fn collect_blob_metadata(&self) -> Result<Vec<(PathBuf, String, fs::Metadata)>, StoreError> {
162        let mut blobs = Vec::new();
163        self.collect_blob_metadata_recursive(&self.base_path, &mut blobs)?;
164        Ok(blobs)
165    }
166
167    /// Sync put operation.
168    pub fn put_sync(&self, hash: Hash, data: &[u8]) -> Result<bool, StoreError> {
169        let path = self.blob_path(&hash);
170
171        // Check if already exists
172        if self.existing_blob_path(&hash).is_some() {
173            return Ok(false);
174        }
175
176        // Create parent directory if needed
177        if let Some(parent) = path.parent() {
178            fs::create_dir_all(parent)?;
179        }
180
181        // Write atomically using temp file + rename
182        let temp_path = path.with_extension("tmp");
183        fs::write(&temp_path, data)?;
184        fs::rename(&temp_path, &path)?;
185
186        Ok(true)
187    }
188
189    /// Sync get operation.
190    pub fn get_sync(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
191        if let Some(path) = self.existing_blob_path(hash) {
192            Ok(Some(fs::read(&path)?))
193        } else {
194            Ok(None)
195        }
196    }
197
198    pub fn get_range_sync(
199        &self,
200        hash: &Hash,
201        start: u64,
202        end_inclusive: u64,
203    ) -> Result<Option<Vec<u8>>, StoreError> {
204        let Some(path) = self.existing_blob_path(hash) else {
205            return Ok(None);
206        };
207
208        let mut file = fs::File::open(path)?;
209        let len = file.metadata()?.len();
210        if len == 0 || start >= len || end_inclusive < start {
211            return Ok(Some(Vec::new()));
212        }
213
214        let actual_end = end_inclusive.min(len - 1);
215        let read_len = actual_end.saturating_sub(start).saturating_add(1);
216        let read_len = usize::try_from(read_len)
217            .map_err(|_| StoreError::Other("blob range is too large to read".to_string()))?;
218        let mut data = vec![0; read_len];
219        file.seek(SeekFrom::Start(start))?;
220        file.read_exact(&mut data)?;
221        Ok(Some(data))
222    }
223
224    pub fn blob_size_sync(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
225        let Some(path) = self.existing_blob_path(hash) else {
226            return Ok(None);
227        };
228        Ok(Some(fs::metadata(path)?.len()))
229    }
230
231    pub fn touch_accessed_sync(&self, _hash: &Hash, _now: u64) -> Result<bool, StoreError> {
232        Ok(false)
233    }
234
235    pub fn touch_many_accessed_sync(
236        &self,
237        _hashes: &[Hash],
238        _now: u64,
239    ) -> Result<usize, StoreError> {
240        Ok(0)
241    }
242
243    pub fn last_accessed_at_sync(&self, hash: &Hash) -> Result<Option<u64>, StoreError> {
244        let Some(path) = self.existing_blob_path(hash) else {
245            return Ok(None);
246        };
247        let metadata = fs::metadata(path)?;
248        Ok(Some(system_time_to_unix_secs(
249            metadata.accessed().or_else(|_| metadata.modified())?,
250        )))
251    }
252
253    pub fn many_last_accessed_at_sync(
254        &self,
255        hashes: &[Hash],
256    ) -> Result<Vec<(Hash, u64)>, StoreError> {
257        let mut results = Vec::new();
258        for hash in hashes {
259            if let Some(last_accessed_at) = self.last_accessed_at_sync(hash)? {
260                results.push((*hash, last_accessed_at));
261            }
262        }
263        Ok(results)
264    }
265
266    /// Check if a hash exists.
267    pub fn exists(&self, hash: &Hash) -> bool {
268        self.existing_blob_path(hash).is_some()
269    }
270
271    /// Sync delete operation.
272    pub fn delete_sync(&self, hash: &Hash) -> Result<bool, StoreError> {
273        let primary = self.blob_path(hash);
274        let legacy = self.legacy_blob_path(hash);
275        let mut deleted = false;
276
277        for path in [primary, legacy] {
278            if path.exists() {
279                fs::remove_file(path)?;
280                deleted = true;
281            }
282        }
283
284        Ok(deleted)
285    }
286
287    /// List all hashes in the store.
288    pub fn list(&self) -> Result<Vec<Hash>, StoreError> {
289        let mut hashes = Vec::new();
290
291        for (_, full_hex, _) in self.collect_blob_metadata()? {
292            if let Ok(bytes) = hex::decode(&full_hex) {
293                if bytes.len() == 32 {
294                    let mut hash = [0u8; 32];
295                    hash.copy_from_slice(&bytes);
296                    hashes.push(hash);
297                }
298            }
299        }
300
301        Ok(hashes)
302    }
303
304    /// Get storage statistics.
305    pub fn stats(&self) -> Result<FsStats, StoreError> {
306        let pins = self.pins.read().unwrap();
307        let mut count = 0usize;
308        let mut total_bytes = 0u64;
309        let mut pinned_count = 0usize;
310        let mut pinned_bytes = 0u64;
311
312        for (_, hex, metadata) in self.collect_blob_metadata()? {
313            let size = metadata.len();
314            count += 1;
315            total_bytes += size;
316
317            if pins.get(&hex).copied().unwrap_or(0) > 0 {
318                pinned_count += 1;
319                pinned_bytes += size;
320            }
321        }
322
323        Ok(FsStats {
324            count,
325            total_bytes,
326            pinned_count,
327            pinned_bytes,
328        })
329    }
330
331    /// Collect all blobs with their mtime and size for eviction
332    fn collect_blobs_for_eviction(&self) -> Vec<(PathBuf, String, SystemTime, u64)> {
333        self.collect_blob_metadata()
334            .map(|blobs| {
335                blobs
336                    .into_iter()
337                    .map(|(path, hex, metadata)| {
338                        let mtime = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH);
339                        let size = metadata.len();
340                        (path, hex, mtime, size)
341                    })
342                    .collect()
343            })
344            .unwrap_or_default()
345    }
346
347    /// Evict unpinned blobs until storage is under target_bytes
348    fn evict_to_target(&self, target_bytes: u64) -> u64 {
349        let pins = self.pins.read().unwrap();
350
351        // Collect all blobs
352        let mut blobs = self.collect_blobs_for_eviction();
353
354        // Filter to unpinned only
355        blobs.retain(|(_, hex, _, _)| pins.get(hex).copied().unwrap_or(0) == 0);
356
357        // Sort by mtime (oldest first)
358        blobs.sort_by_key(|(_, _, mtime, _)| *mtime);
359
360        drop(pins); // Release lock before deleting
361
362        // Calculate current total
363        let current_bytes: u64 = self
364            .collect_blobs_for_eviction()
365            .iter()
366            .map(|(_, _, _, size)| *size)
367            .sum();
368
369        if current_bytes <= target_bytes {
370            return 0;
371        }
372
373        let to_free = current_bytes - target_bytes;
374        let mut freed = 0u64;
375
376        for (path, _, _, size) in blobs {
377            if freed >= to_free {
378                break;
379            }
380            if fs::remove_file(&path).is_ok() {
381                freed += size;
382            }
383        }
384
385        freed
386    }
387}
388
389fn system_time_to_unix_secs(time: SystemTime) -> u64 {
390    time.duration_since(UNIX_EPOCH)
391        .unwrap_or_default()
392        .as_secs()
393}
394
395/// Storage statistics.
396#[derive(Debug, Clone)]
397pub struct FsStats {
398    pub count: usize,
399    pub total_bytes: u64,
400    pub pinned_count: usize,
401    pub pinned_bytes: u64,
402}
403
404#[async_trait]
405impl Store for FsBlobStore {
406    async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError> {
407        self.put_sync(hash, &data)
408    }
409
410    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
411        self.get_sync(hash)
412    }
413
414    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
415        Ok(self.exists(hash))
416    }
417
418    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
419        let hex = hex::encode(hash);
420        // Remove pin entry if exists
421        {
422            let mut pins = self.pins.write().unwrap();
423            pins.remove(&hex);
424        }
425        let _ = self.save_pins(); // Best effort
426        self.delete_sync(hash)
427    }
428
429    fn set_max_bytes(&self, max: u64) {
430        self.max_bytes.store(max, Ordering::Relaxed);
431    }
432
433    fn max_bytes(&self) -> Option<u64> {
434        let max = self.max_bytes.load(Ordering::Relaxed);
435        if max > 0 {
436            Some(max)
437        } else {
438            None
439        }
440    }
441
442    async fn stats(&self) -> StoreStats {
443        match self.stats() {
444            Ok(fs_stats) => StoreStats {
445                count: fs_stats.count as u64,
446                bytes: fs_stats.total_bytes,
447                pinned_count: fs_stats.pinned_count as u64,
448                pinned_bytes: fs_stats.pinned_bytes,
449            },
450            Err(_) => StoreStats::default(),
451        }
452    }
453
454    async fn evict_if_needed(&self) -> Result<u64, StoreError> {
455        let max = self.max_bytes.load(Ordering::Relaxed);
456        if max == 0 {
457            return Ok(0); // No limit set
458        }
459
460        let current = match self.stats() {
461            Ok(s) => s.total_bytes,
462            Err(_) => return Ok(0),
463        };
464
465        if current <= max {
466            return Ok(0);
467        }
468
469        // Evict to 90% of max
470        let target = max * 9 / 10;
471        Ok(self.evict_to_target(target))
472    }
473
474    async fn pin(&self, hash: &Hash) -> Result<(), StoreError> {
475        let hex = hex::encode(hash);
476        {
477            let mut pins = self.pins.write().unwrap();
478            *pins.entry(hex).or_insert(0) += 1;
479        }
480        self.save_pins()
481    }
482
483    async fn unpin(&self, hash: &Hash) -> Result<(), StoreError> {
484        let hex = hex::encode(hash);
485        {
486            let mut pins = self.pins.write().unwrap();
487            if let Some(count) = pins.get_mut(&hex) {
488                if *count > 0 {
489                    *count -= 1;
490                }
491                if *count == 0 {
492                    pins.remove(&hex);
493                }
494            }
495        }
496        self.save_pins()
497    }
498
499    fn pin_count(&self, hash: &Hash) -> u32 {
500        let hex = hex::encode(hash);
501        self.pins.read().unwrap().get(&hex).copied().unwrap_or(0)
502    }
503}
504
505#[cfg(test)]
506mod tests {
507    use super::*;
508    use hashtree_core::sha256;
509    use tempfile::TempDir;
510
511    #[tokio::test]
512    async fn test_put_get() {
513        let temp = TempDir::new().unwrap();
514        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
515
516        let data = b"hello filesystem";
517        let hash = sha256(data);
518        store.put(hash, data.to_vec()).await.unwrap();
519
520        assert!(store.has(&hash).await.unwrap());
521        assert_eq!(store.get(&hash).await.unwrap(), Some(data.to_vec()));
522    }
523
524    #[tokio::test]
525    async fn test_get_missing() {
526        let temp = TempDir::new().unwrap();
527        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
528
529        let hash = [0u8; 32];
530        assert!(!store.has(&hash).await.unwrap());
531        assert_eq!(store.get(&hash).await.unwrap(), None);
532    }
533
534    #[tokio::test]
535    async fn test_delete() {
536        let temp = TempDir::new().unwrap();
537        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
538
539        let data = b"delete me";
540        let hash = sha256(data);
541        store.put(hash, data.to_vec()).await.unwrap();
542        assert!(store.has(&hash).await.unwrap());
543
544        assert!(store.delete(&hash).await.unwrap());
545        assert!(!store.has(&hash).await.unwrap());
546        assert!(!store.delete(&hash).await.unwrap());
547    }
548
549    #[tokio::test]
550    async fn test_deduplication() {
551        let temp = TempDir::new().unwrap();
552        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
553
554        let data = b"same content";
555        let hash = sha256(data);
556
557        // First put returns true (newly stored)
558        assert!(store.put(hash, data.to_vec()).await.unwrap());
559        // Second put returns false (already existed)
560        assert!(!store.put(hash, data.to_vec()).await.unwrap());
561
562        assert_eq!(store.list().unwrap().len(), 1);
563    }
564
565    #[tokio::test]
566    async fn test_list() {
567        let temp = TempDir::new().unwrap();
568        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
569
570        let d1 = b"one";
571        let d2 = b"two";
572        let d3 = b"three";
573        let h1 = sha256(d1);
574        let h2 = sha256(d2);
575        let h3 = sha256(d3);
576
577        store.put(h1, d1.to_vec()).await.unwrap();
578        store.put(h2, d2.to_vec()).await.unwrap();
579        store.put(h3, d3.to_vec()).await.unwrap();
580
581        let hashes = store.list().unwrap();
582        assert_eq!(hashes.len(), 3);
583        assert!(hashes.contains(&h1));
584        assert!(hashes.contains(&h2));
585        assert!(hashes.contains(&h3));
586    }
587
588    #[tokio::test]
589    async fn test_stats() {
590        let temp = TempDir::new().unwrap();
591        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
592
593        let d1 = b"hello";
594        let d2 = b"world";
595        let h1 = sha256(d1);
596        store.put(h1, d1.to_vec()).await.unwrap();
597        store.put(sha256(d2), d2.to_vec()).await.unwrap();
598
599        let stats = store.stats().unwrap();
600        assert_eq!(stats.count, 2);
601        assert_eq!(stats.total_bytes, 10);
602        assert_eq!(stats.pinned_count, 0);
603        assert_eq!(stats.pinned_bytes, 0);
604
605        // Pin one item and check stats
606        store.pin(&h1).await.unwrap();
607        let stats = store.stats().unwrap();
608        assert_eq!(stats.pinned_count, 1);
609        assert_eq!(stats.pinned_bytes, 5);
610    }
611
612    #[tokio::test]
613    async fn test_directory_structure() {
614        let temp = TempDir::new().unwrap();
615        let blobs_path = temp.path().join("blobs");
616        let store = FsBlobStore::new(&blobs_path).unwrap();
617
618        let data = b"test data";
619        let hash = sha256(data);
620        let hex = hex::encode(hash);
621
622        store.put(hash, data.to_vec()).await.unwrap();
623
624        // Verify the file exists at the correct path
625        let prefix = &hex[..2];
626        let subdir = &hex[2..4];
627        let rest = &hex[4..];
628        let expected_path = blobs_path.join(prefix).join(subdir).join(rest);
629
630        assert!(
631            expected_path.exists(),
632            "Blob should be at {:?}",
633            expected_path
634        );
635        assert_eq!(fs::read(&expected_path).unwrap(), data);
636    }
637
638    #[test]
639    fn test_blob_path_format() {
640        let temp = TempDir::new().unwrap();
641        let store = FsBlobStore::new(temp.path()).unwrap();
642
643        // Hash: 0x00112233...
644        let mut hash = [0u8; 32];
645        hash[0] = 0x00;
646        hash[1] = 0x11;
647        hash[2] = 0x22;
648
649        let path = store.blob_path(&hash);
650        let path_str = path.to_string_lossy();
651
652        // Should have "00" as directory prefix
653        assert!(
654            path_str.contains("/00/"),
655            "Path should contain /00/ directory: {}",
656            path_str
657        );
658        // Should also include a second shard level based on the next byte.
659        assert!(
660            path_str.contains("/11/"),
661            "Path should contain /11/ directory: {}",
662            path_str
663        );
664        // File name should be remaining 60 chars
665        assert!(path.file_name().unwrap().len() == 60);
666    }
667
668    #[tokio::test]
669    async fn test_legacy_single_level_layout_remains_readable() {
670        let temp = TempDir::new().unwrap();
671        let blobs_path = temp.path().join("blobs");
672        let store = FsBlobStore::new(&blobs_path).unwrap();
673
674        let data = b"legacy blob";
675        let hash = sha256(data);
676        let hex = hex::encode(hash);
677        let legacy_path = blobs_path.join(&hex[..2]).join(&hex[2..]);
678        fs::create_dir_all(legacy_path.parent().unwrap()).unwrap();
679        fs::write(&legacy_path, data).unwrap();
680
681        assert!(store.has(&hash).await.unwrap());
682        assert_eq!(store.get(&hash).await.unwrap(), Some(data.to_vec()));
683
684        let listed = store.list().unwrap();
685        assert_eq!(listed, vec![hash]);
686
687        let stats = store.stats().unwrap();
688        assert_eq!(stats.count, 1);
689        assert_eq!(stats.total_bytes, data.len() as u64);
690
691        assert!(!store.put(hash, data.to_vec()).await.unwrap());
692        assert!(store.delete(&hash).await.unwrap());
693        assert!(!legacy_path.exists());
694    }
695
696    #[tokio::test]
697    async fn test_empty_store_stats() {
698        let temp = TempDir::new().unwrap();
699        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
700
701        let stats = store.stats().unwrap();
702        assert_eq!(stats.count, 0);
703        assert_eq!(stats.total_bytes, 0);
704    }
705
706    #[tokio::test]
707    async fn test_empty_store_list() {
708        let temp = TempDir::new().unwrap();
709        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
710
711        let hashes = store.list().unwrap();
712        assert!(hashes.is_empty());
713    }
714
715    #[tokio::test]
716    async fn test_pin_and_unpin() {
717        let temp = TempDir::new().unwrap();
718        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
719
720        let data = b"pin me";
721        let hash = sha256(data);
722        store.put(hash, data.to_vec()).await.unwrap();
723
724        // Initially not pinned
725        assert!(!store.is_pinned(&hash));
726        assert_eq!(store.pin_count(&hash), 0);
727
728        // Pin
729        store.pin(&hash).await.unwrap();
730        assert!(store.is_pinned(&hash));
731        assert_eq!(store.pin_count(&hash), 1);
732
733        // Unpin
734        store.unpin(&hash).await.unwrap();
735        assert!(!store.is_pinned(&hash));
736        assert_eq!(store.pin_count(&hash), 0);
737    }
738
739    #[tokio::test]
740    async fn test_pin_ref_counting() {
741        let temp = TempDir::new().unwrap();
742        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
743
744        let data = b"multi pin";
745        let hash = sha256(data);
746        store.put(hash, data.to_vec()).await.unwrap();
747
748        // Pin multiple times
749        store.pin(&hash).await.unwrap();
750        store.pin(&hash).await.unwrap();
751        store.pin(&hash).await.unwrap();
752        assert_eq!(store.pin_count(&hash), 3);
753
754        // Unpin once
755        store.unpin(&hash).await.unwrap();
756        assert_eq!(store.pin_count(&hash), 2);
757        assert!(store.is_pinned(&hash));
758
759        // Unpin remaining
760        store.unpin(&hash).await.unwrap();
761        store.unpin(&hash).await.unwrap();
762        assert_eq!(store.pin_count(&hash), 0);
763    }
764
765    #[tokio::test]
766    async fn test_pins_persist_across_reload() {
767        let temp = TempDir::new().unwrap();
768        let blobs_path = temp.path().join("blobs");
769
770        let data = b"persist me";
771        let hash = sha256(data);
772
773        // Create store and pin
774        {
775            let store = FsBlobStore::new(&blobs_path).unwrap();
776            store.put(hash, data.to_vec()).await.unwrap();
777            store.pin(&hash).await.unwrap();
778            store.pin(&hash).await.unwrap();
779            assert_eq!(store.pin_count(&hash), 2);
780        }
781
782        // Reload store
783        {
784            let store = FsBlobStore::new(&blobs_path).unwrap();
785            assert_eq!(store.pin_count(&hash), 2);
786            assert!(store.is_pinned(&hash));
787        }
788    }
789
790    #[tokio::test]
791    async fn test_max_bytes() {
792        let temp = TempDir::new().unwrap();
793        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
794
795        assert!(store.max_bytes().is_none());
796
797        store.set_max_bytes(1000);
798        assert_eq!(store.max_bytes(), Some(1000));
799
800        store.set_max_bytes(0);
801        assert!(store.max_bytes().is_none());
802    }
803
804    #[tokio::test]
805    async fn test_with_max_bytes() {
806        let temp = TempDir::new().unwrap();
807        let store = FsBlobStore::with_max_bytes(temp.path().join("blobs"), 500).unwrap();
808        assert_eq!(store.max_bytes(), Some(500));
809    }
810
811    #[tokio::test]
812    async fn test_eviction_respects_pins() {
813        let temp = TempDir::new().unwrap();
814        // 20 byte limit
815        let store = FsBlobStore::with_max_bytes(temp.path().join("blobs"), 20).unwrap();
816
817        // Add items (5 bytes each = 15 total)
818        let d1 = b"aaaaa"; // oldest - will be pinned
819        let d2 = b"bbbbb";
820        let d3 = b"ccccc";
821        let h1 = sha256(d1);
822        let h2 = sha256(d2);
823        let h3 = sha256(d3);
824
825        store.put(h1, d1.to_vec()).await.unwrap();
826        std::thread::sleep(std::time::Duration::from_millis(10)); // Ensure different mtime
827        store.put(h2, d2.to_vec()).await.unwrap();
828        std::thread::sleep(std::time::Duration::from_millis(10));
829        store.put(h3, d3.to_vec()).await.unwrap();
830
831        // Pin the oldest
832        store.pin(&h1).await.unwrap();
833
834        // Add more to exceed limit (15 + 5 = 20, at limit)
835        let d4 = b"ddddd";
836        let h4 = sha256(d4);
837        std::thread::sleep(std::time::Duration::from_millis(10));
838        store.put(h4, d4.to_vec()).await.unwrap();
839
840        // Add one more to exceed (20 + 5 = 25 > 20)
841        let d5 = b"eeeee";
842        let h5 = sha256(d5);
843        std::thread::sleep(std::time::Duration::from_millis(10));
844        store.put(h5, d5.to_vec()).await.unwrap();
845
846        // Evict
847        let freed = store.evict_if_needed().await.unwrap();
848        assert!(freed > 0, "Should have freed some bytes");
849
850        // Pinned item should still exist
851        assert!(store.has(&h1).await.unwrap(), "Pinned item should exist");
852        // Oldest unpinned (h2) should be evicted
853        assert!(
854            !store.has(&h2).await.unwrap(),
855            "Oldest unpinned should be evicted"
856        );
857        // Newest should exist
858        assert!(store.has(&h5).await.unwrap(), "Newest should exist");
859    }
860
861    #[tokio::test]
862    async fn test_no_eviction_when_under_limit() {
863        let temp = TempDir::new().unwrap();
864        let store = FsBlobStore::with_max_bytes(temp.path().join("blobs"), 1000).unwrap();
865
866        let data = b"small";
867        let hash = sha256(data);
868        store.put(hash, data.to_vec()).await.unwrap();
869
870        let freed = store.evict_if_needed().await.unwrap();
871        assert_eq!(freed, 0);
872        assert!(store.has(&hash).await.unwrap());
873    }
874
875    #[tokio::test]
876    async fn test_no_eviction_without_limit() {
877        let temp = TempDir::new().unwrap();
878        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
879
880        for i in 0..10u8 {
881            let data = vec![i; 100];
882            let hash = sha256(&data);
883            store.put(hash, data).await.unwrap();
884        }
885
886        let freed = store.evict_if_needed().await.unwrap();
887        assert_eq!(freed, 0);
888        assert_eq!(store.list().unwrap().len(), 10);
889    }
890
891    #[tokio::test]
892    async fn test_delete_removes_pin() {
893        let temp = TempDir::new().unwrap();
894        let store = FsBlobStore::new(temp.path().join("blobs")).unwrap();
895
896        let data = b"delete pinned";
897        let hash = sha256(data);
898        store.put(hash, data.to_vec()).await.unwrap();
899        store.pin(&hash).await.unwrap();
900        assert!(store.is_pinned(&hash));
901
902        store.delete(&hash).await.unwrap();
903        assert_eq!(store.pin_count(&hash), 0);
904    }
905}